test(e2e): give concurrent TLS reads independent tasks

crates/concurrency/Cargo.toml

@@ -26,7 +26,7 @@ rustfs-io-core = { workspace = true }
 serde = { workspace = true, features = ["derive"] }
 
 # Async runtime
-tokio = { workspace = true, features = ["sync", "fs", "rt-multi-thread"] }
+tokio = { workspace = true, features = ["sync", "fs", "rt-multi-thread", "time"] }
 
 # Logging
 tracing = { workspace = true }

crates/e2e_test/src/common.rs

@@ -103,6 +103,18 @@ fn capture_log_path(log_dir: &Path, temp_dir: &str) -> Option<PathBuf> {
     Some(log_dir.join(format!("{temp_name}.log")))
 }
 
+#[cfg(test)]
+pub(crate) fn tls_diagnostic_log_directory(report: Option<&Path>, artifacts: Option<&Path>, cluster: &Path) -> PathBuf {
+    if let Some(report) = report {
+        return report.with_extension("logs");
+    }
+    if let Some(artifacts) = artifacts {
+        // Keep failed-run logs outside the cluster directory removed on drop.
+        return artifacts.join(format!("internode-tls-{}", uuid::Uuid::new_v4().simple()));
+    }
+    cluster.join("logs")
+}
+
 fn configured_capture_log_path(temp_dir: &str) -> Option<String> {
     let log_dir = std::env::var_os("RUSTFS_E2E_LOG_DIR")?;
     if stdfs::create_dir_all(&log_dir).is_err() {
@@ -2319,6 +2331,25 @@ mod tests {
             capture_log_path(Path::new("/tmp/e2e-logs"), "/tmp/rustfs_e2e_test_abc"),
             Some(PathBuf::from("/tmp/e2e-logs/rustfs_e2e_test_abc.log"))
         );
+
+        let cluster = tempfile::tempdir().expect("temporary cluster");
+        let artifacts = tempfile::tempdir().expect("CI artifact root");
+        let first = tls_diagnostic_log_directory(None, Some(artifacts.path()), cluster.path());
+        let second = tls_diagnostic_log_directory(None, Some(artifacts.path()), cluster.path());
+        assert!(first.starts_with(artifacts.path()), "CI must retain TLS node logs after cluster cleanup");
+        assert!(second.starts_with(artifacts.path()));
+        assert_ne!(first, second, "separate attempts must not append to another run's logs");
+        assert_eq!(
+            tls_diagnostic_log_directory(Some(Path::new("acceptance/report.json")), Some(artifacts.path()), cluster.path()),
+            Path::new("acceptance/report.logs"),
+            "an explicit acceptance report keeps its existing node-log location"
+        );
+        assert_eq!(tls_diagnostic_log_directory(None, None, cluster.path()), cluster.path().join("logs"));
+        std::fs::create_dir(&first).expect("run log directory");
+        let log = first.join("node-0.log");
+        std::fs::write(&log, "read failure").expect("captured node output");
+        cluster.close().expect("failed cluster cleanup");
+        assert_eq!(std::fs::read_to_string(log).expect("retained node log"), "read failure");
     }
 
     #[test]

crates/e2e_test/src/distributed/internode_tls_test.rs

@@ -2,14 +2,16 @@
 // Licensed under the Apache License, Version 2.0.
 
 use super::harness::{DistCluster, DistLayout, TestResult, unique_bucket};
-use crate::tls_gen;
+use crate::{common::tls_diagnostic_log_directory, tls_gen};
+use anyhow::Context;
 use aws_sdk_s3::{Client, config::retry::RetryConfig, primitives::ByteStream};
-use futures::future::try_join_all;
 use rand::{Rng, SeedableRng, rngs::StdRng};
 use sha2::{Digest, Sha256};
 use std::io::{Read, Seek, SeekFrom, Write};
 use std::path::{Path, PathBuf};
+use std::sync::Arc;
 use std::time::{Duration, Instant};
+use tokio::task::JoinSet;
 
 const MIB: u64 = 1024 * 1024;
 const READ_TIMEOUT: Duration = Duration::from_secs(180);
@@ -85,6 +87,8 @@ async fn verified_get(
     expected: &ExpectedRead,
     concurrent: bool,
 ) -> TestResult<serde_json::Value> {
+    let mut stage = "response headers";
+    let mut count = 0u64;
     tokio::time::timeout(READ_TIMEOUT, async {
         let started = Instant::now();
         let mut request = client.get_object().bucket(bucket).key(&object.key);
@@ -92,13 +96,13 @@ async fn verified_get(
             request = request.range(format!("bytes={start}-{end}"));
         }
         let mut output = request.send().await?;
+        stage = "response body";
         assert_eq!(output.content_length(), Some(i64::try_from(expected.length)?));
         if let Some((start, end)) = expected.range {
             assert_eq!(output.content_range(), Some(format!("bytes {start}-{end}/{}", object.length).as_str()));
         } else {
             assert_eq!(output.content_range(), None);
         }
-        let mut count = 0u64;
         let mut digest = Sha256::new();
         while let Some(chunk) = output.body.next().await {
             let chunk = chunk?;
@@ -115,7 +119,16 @@ async fn verified_get(
             "concurrent": concurrent, "elapsed_ms": started.elapsed().as_secs_f64() * 1000.0,
         }))
     })
-    .await?
+    .await
+    .with_context(|| {
+        format!(
+            "TLS GET timed out after {}s: node={node}, key={}, range={:?}, concurrent={concurrent}, stage={stage}, received={count}, expected={}",
+            READ_TIMEOUT.as_secs(),
+            object.key,
+            expected.range,
+            expected.length
+        )
+    })?
 }
 
 /// This case always exercises the full acceptance sizes; do not substitute
@@ -157,9 +170,8 @@ async fn four_node_tls_large_and_range_gets_preserve_bytes() -> TestResult {
     {
         return Err("acceptance report paths must be unused".into());
     }
-    let log_directory = report_path
-        .as_ref()
-        .map_or_else(|| directory.join("logs"), |path| path.with_extension("logs"));
+    let artifacts = std::env::var_os("RUSTFS_E2E_LOG_DIR").map(PathBuf::from);
+    let log_directory = tls_diagnostic_log_directory(report_path.as_deref(), artifacts.as_deref(), &directory);
     std::fs::create_dir(&log_directory)?;
     dist.capture_logs_under(&log_directory)?;
     dist.cluster.start().await?;
@@ -183,7 +195,7 @@ async fn four_node_tls_large_and_range_gets_preserve_bytes() -> TestResult {
     let mut results = Vec::new();
     for length in [32 * 1024, 100 * MIB, 1024 * MIB] {
         let fixture_directory = directory.clone();
-        let object = tokio::task::spawn_blocking(move || fixture(&fixture_directory, length)).await??;
+        let object = Arc::new(tokio::task::spawn_blocking(move || fixture(&fixture_directory, length)).await??);
         let started = Instant::now();
         tokio::time::timeout(
             READ_TIMEOUT,
@@ -194,7 +206,14 @@ async fn four_node_tls_large_and_range_gets_preserve_bytes() -> TestResult {
                 .body(ByteStream::from_path(&object.path).await?)
                 .send(),
         )
-        .await??;
+        .await
+        .with_context(|| {
+            format!(
+                "TLS PUT timed out after {}s: node=0, key={}, bytes={length}",
+                READ_TIMEOUT.as_secs(),
+                object.key
+            )
+        })??;
         results.push(
             serde_json::json!({"operation": "put", "bytes": length, "elapsed_ms": started.elapsed().as_secs_f64() * 1000.0}),
         );
@@ -216,15 +235,18 @@ async fn four_node_tls_large_and_range_gets_preserve_bytes() -> TestResult {
                 results.push(verified_get(client, node, &bucket, &object, expected, false).await?);
             }
         }
-        results.extend(
-            try_join_all(
-                clients
-                    .iter()
-                    .enumerate()
-                    .map(|(node, client)| verified_get(client, node, &bucket, &object, &object.reads[0], true)),
-            )
-            .await?,
-        );
+        // Ready body chunks can exhaust one task's cooperative budget before
+        // its sibling futures are polled. Each GET needs its own budget.
+        let mut concurrent_gets = JoinSet::new();
+        for (node, client) in clients.iter().enumerate() {
+            let client = client.clone();
+            let bucket = bucket.clone();
+            let object = object.clone();
+            concurrent_gets.spawn(async move { verified_get(&client, node, &bucket, &object, &object.reads[0], true).await });
+        }
+        while let Some(result) = concurrent_gets.join_next().await {
+            results.push(result??);
+        }
         tokio::fs::remove_file(&object.path).await?;
     }
     for node in 0..clients.len() {

crates/scanner/src/error.rs

@@ -48,6 +48,10 @@ pub enum ScannerError {
     #[error("Remote namespace scanner disk is already active")]
     RemoteDiskBusy,
 
+    /// Node resource pressure deferred work before its first storage poll.
+    #[error("Scanner node resource admission deferred")]
+    ResourceDeferred,
+
     /// Partial data usage cache produced before the scanner stopped.
     #[error("Scanner stopped with partial data usage cache")]
     PartialCache(Box<DataUsageCache>),

crates/scanner/src/lib.rs

@@ -123,7 +123,7 @@ pub use storage_api::scan::{
     SCANNER_DIRTY_USAGE_SNAPSHOT_PROTOCOL_VERSION, SCANNER_DIRTY_USAGE_SNAPSHOT_RPC_MAX_MESSAGE_SIZE,
     SCANNER_SCOPED_DIRTY_USAGE_ACK_MAX_ENTRIES,
 };
-pub use workload_admission::set_scanner_workload_admission_snapshot_provider;
+pub use workload_admission::{set_scanner_node_admission, set_scanner_workload_admission_snapshot_provider};
 
 static SCANNER_ACTIVE_WORK_UNITS: AtomicU64 = AtomicU64::new(0);
 static SCANNER_RUNTIME_INSTANCES: AtomicU64 = AtomicU64::new(0);

crates/scanner/src/remote_scanner/stream.rs

@@ -705,7 +705,7 @@ async fn scan_and_persist_local_bucket(
         RemoteScannerServerError::worker("remote namespace scanner cache publication is blocked by data movement")
     })?;
     let cache_name = path_join_buf(&[&bucket, DATA_USAGE_CACHE_NAME]);
-    let guard = acquire_scanner_cache_locks(set.as_ref(), &cache_name, source)
+    let mut guard = acquire_scanner_cache_locks(set.as_ref(), &cache_name, source)
         .await
         .map_err(|err| {
             if err.is_contention() {
@@ -791,19 +791,23 @@ async fn scan_and_persist_local_bucket(
 
     let set_disks = scanner_set_disk_inventory(set.as_ref()).await;
     let (checkpoint_tx, mut checkpoint_rx) = mpsc::channel::<DataUsageCache>(1);
+    let (node_tx, mut node_rx) = crate::workload_admission::scanner_node_request_channel();
     let scan_ctx = ctx.child_token();
-    let scan = ScannerIODisk::nsscanner_disk(
-        disk.clone(),
-        scan_ctx.clone(),
-        budget,
-        set_disks,
-        cache,
-        None,
-        ScannerDiskScanOptions {
-            scan_mode,
-            prefix_scan_scope: None,
-            checkpoint_tx: Some(checkpoint_tx),
-        },
+    let scan = crate::workload_admission::with_scanner_node_requests(
+        node_tx,
+        ScannerIODisk::nsscanner_disk(
+            disk.clone(),
+            scan_ctx.clone(),
+            budget,
+            set_disks,
+            cache,
+            None,
+            ScannerDiskScanOptions {
+                scan_mode,
+                prefix_scan_scope: None,
+                checkpoint_tx: Some(checkpoint_tx),
+            },
+        ),
     );
     tokio::pin!(scan);
     let fence_watch = watch_remote_scanner_request_fence(next_cycle, leader_epoch, store.clone(), NS_SCANNER_FENCE_POLL_INTERVAL);
@@ -813,6 +817,27 @@ async fn scan_and_persist_local_bucket(
     let mut checkpoint_channel_closed = false;
     let outcome = loop {
         tokio::select! {
+            request = async {
+                match node_rx.as_mut() {
+                    Some(receiver) => receiver.recv().await,
+                    None => std::future::pending().await,
+                }
+            } => {
+                let Some(request) = request else {
+                    node_rx = None;
+                    continue;
+                };
+                if let Err(error) = wait_for_remote_node_resumption(
+                    guard.resume_node_operation(set.clone(), &cache_name, source, &revisions, request),
+                    fence_watch.as_mut(),
+                ).await {
+                    scan_ctx.cancel();
+                    let _ = tokio::time::timeout(NS_SCANNER_LOCK_LOSS_SHUTDOWN_TIMEOUT, scan.as_mut()).await;
+                    return Err(RemoteScannerServerError::retry_bucket(format!(
+                        "remote namespace scanner could not resume node admission: {error}",
+                    )));
+                }
+            }
             result = &mut scan => {
                 match result {
                     Ok(outcome) => break outcome,
@@ -1121,6 +1146,22 @@ pub(crate) async fn scan_remote_bucket(
     result
 }
 
+/// Resource waits must keep checking the persisted leader fence.
+async fn wait_for_remote_node_resumption<F, G>(resume: F, fence_watch: Pin<&mut G>) -> Result<(), StorageError>
+where
+    F: Future<Output = Result<(), StorageError>>,
+    G: Future<Output = Result<(), ScannerError>>,
+{
+    tokio::select! {
+        biased;
+        result = fence_watch => Err(match result {
+            Ok(()) => StorageError::other("remote namespace scanner fence watcher stopped unexpectedly"),
+            Err(error) => StorageError::other(error),
+        }),
+        result = resume => result,
+    }
+}
+
 fn finish_remote_scanner_stream(
     result: RemoteScannerStreamResult<RemoteScannerOutcome>,
     budget: &ScannerCycleBudget,

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

@@ -906,6 +906,58 @@ async fn fence_watcher_stops_work_after_persisted_leader_changes() {
     assert!(err.to_string().contains("requested_epoch=9"));
 }
 
+#[tokio::test]
+async fn node_admission_wait_does_not_hide_replacement_leader() {
+    use rustfs_concurrency::adaptive::{FeedbackController, FeedbackObservation, NodeAdmission, ResourcePressure};
+    use std::num::NonZeroU16;
+
+    let store = Arc::new(MutableCycleStateStore {
+        state: RwLock::new(crate::scanner::encode_scanner_cycle_fence_for_test(7, 9)),
+    });
+    let cache = Mutex::new(None);
+    let refresh = AsyncMutex::new(());
+    let watcher = watch_remote_scanner_request_fence_with_cache(
+        7,
+        9,
+        store.clone(),
+        Duration::from_millis(5),
+        &cache,
+        &refresh,
+        Duration::from_millis(5),
+    );
+    tokio::pin!(watcher);
+    let gate = NodeAdmission::new(NonZeroU16::new(1).expect("one shared node unit"));
+    let storage_polled = AtomicBool::new(false);
+    {
+        let resume = async {
+            let _permit = gate.acquire().await;
+            storage_polled.store(true, Ordering::Relaxed);
+            Ok(())
+        };
+        let waiting = wait_for_remote_node_resumption(resume, watcher.as_mut());
+        tokio::pin!(waiting);
+        assert!(futures::poll!(&mut waiting).is_pending(), "no resource sample grants a node unit");
+        *store.state.write().expect("persist replacement leader") = crate::scanner::encode_scanner_cycle_fence_for_test(7, 10);
+        let error = tokio::time::timeout(Duration::from_millis(100), waiting)
+            .await
+            .expect("resource pause must not hide a changed leader")
+            .expect_err("old leader cannot resume storage work");
+        assert!(error.to_string().contains("requested_epoch=9"));
+    }
+    assert!(!storage_polled.load(Ordering::Relaxed));
+    assert_eq!(gate.snapshot().active, 0);
+    gate.update(
+        &mut FeedbackController::new(),
+        Some(FeedbackObservation {
+            generation: 1,
+            max_concurrency: 1,
+            pressure: ResourcePressure::Unknown,
+            completed_units: None,
+        }),
+    );
+    assert!(gate.try_acquire().is_some(), "failed fencing drops the old node queue ticket");
+}
+
 #[tokio::test]
 async fn concurrent_cycle_cache_misses_share_one_backend_read() {
     let request = RemoteScannerRequest(test_request(Uuid::new_v4()));

crates/scanner/src/scanner_folder.rs

@@ -42,6 +42,7 @@ use crate::storage_api::owner::{
 };
 #[cfg(test)]
 use crate::storage_api::owner::{EcstoreExpirationStatus as ExpirationStatus, EcstoreLifecycleRule as LifecycleRule};
+use crate::workload_admission::run_scanner_node_operation;
 use metrics::{counter, describe_counter};
 use rustfs_common::trace_bus::{TraceEvent, TraceFunc, TraceKind, trace_emit, trace_subscriber_count};
 use rustfs_filemeta::{
@@ -769,6 +770,7 @@ pub struct FolderScanner {
     resolved_failed_object_paths: HashSet<String>,
 
     sleeper: DynamicSleeper,
+    node_admission: Option<Arc<rustfs_concurrency::adaptive::NodeAdmission>>,
     // should_heal: Arc<dyn Fn() -> bool + Send + Sync>,
     disks: Vec<Arc<Disk>>,
     disks_quorum: usize,
@@ -807,6 +809,8 @@ pub struct FolderScanner {
     pending_size_reconciliation_truncated: bool,
     #[cfg(test)]
     list_path_raw_options_observer: Option<mpsc::UnboundedSender<ListPathRawTimeoutSnapshot>>,
+    #[cfg(test)]
+    defer_abandoned_entries_until_finished: bool,
 }
 
 struct RawEnumerationProgress {
@@ -977,12 +981,13 @@ async fn next_scanner_dir_entry(
     ctx: &CancellationToken,
     sleeper: &DynamicSleeper,
     entries: &mut ScannerDirectory,
+    admission: Option<&Arc<rustfs_concurrency::adaptive::NodeAdmission>>,
 ) -> Result<Option<tokio::fs::DirEntry>, ScannerError> {
     if entries.batch_entries == SCANNER_DIRECTORY_BATCH_ENTRIES {
         wait_for_scanner_io(ctx, tokio::task::yield_now()).await?;
         entries.batch_entries = 0;
     }
-    let entry = wait_for_scanner_io(ctx, entries.next_entry()).await??;
+    let entry = wait_for_scanner_io(ctx, run_scanner_node_operation(admission, ctx, entries.next_entry())).await???;
     if entry.is_some() {
         // Tokio prefetches bounded chunks. Pacing consumption also paces the
         // next refill, without charging an extra entry for EOF.
@@ -1861,7 +1866,12 @@ impl FolderScanner {
             // directory that vanished during iteration instead of failing the
             // whole parent subtree. The bucket root (no parent) still propagates
             // NotFound so namespace removal is detected by the caller.
-            let mut dir_reader = match wait_for_scanner_io(&ctx, ScannerDirectory::open(&dir_path)).await? {
+            let mut dir_reader = match wait_for_scanner_io(
+                &ctx,
+                run_scanner_node_operation(self.node_admission.as_ref(), &ctx, ScannerDirectory::open(&dir_path)),
+            )
+            .await??
+            {
                 Ok(dir_reader) => Some(dir_reader),
                 Err(e)
                     if folder.parent.is_some()
@@ -1899,7 +1909,7 @@ impl FolderScanner {
                 // Hold at most one metadata entry until a candidate reaches
                 // EOF: a later part must not leave object accounting or ILM work.
                 let (entry, replayed_metadata) = if let Some(reader) = dir_reader.as_mut() {
-                    let entry = match next_scanner_dir_entry(&ctx, &self.sleeper, reader).await {
+                    let entry = match next_scanner_dir_entry(&ctx, &self.sleeper, reader, self.node_admission.as_ref()).await {
                         Ok(Some(entry)) => entry,
                         Ok(None) => {
                             raw_enumeration_complete = true;
@@ -2010,7 +2020,12 @@ impl FolderScanner {
 
                 // Ignore entries that disappeared during traversal or hit symlink
                 // loops, but propagate other walk errors.
-                let mut entry_type = match wait_for_scanner_io(&ctx, entry.file_type()).await? {
+                let mut entry_type = match wait_for_scanner_io(
+                    &ctx,
+                    run_scanner_node_operation(self.node_admission.as_ref(), &ctx, entry.file_type()),
+                )
+                .await??
+                {
                     Ok(entry_type) => entry_type,
                     Err(e) if e.kind() == ErrorKind::NotFound => {
                         debug!(
@@ -2061,7 +2076,12 @@ impl FolderScanner {
                 let mut entry_metadata_len = None;
                 if entry_type.is_symlink() {
                     wait_for_scanner_io(&ctx, self.sleeper.admit_disk_io(1, 0)).await?;
-                    let metadata = match wait_for_scanner_io(&ctx, tokio::fs::metadata(&file_path)).await? {
+                    let metadata = match wait_for_scanner_io(
+                        &ctx,
+                        run_scanner_node_operation(self.node_admission.as_ref(), &ctx, tokio::fs::metadata(&file_path)),
+                    )
+                    .await??
+                    {
                         Ok(metadata) => metadata,
                         Err(e) if e.kind() == ErrorKind::NotFound => {
                             debug!(
@@ -2203,21 +2223,26 @@ impl FolderScanner {
                         len
                     } else {
                         wait_for_scanner_io(&ctx, self.sleeper.admit_disk_io(1, 0)).await?;
-                        wait_for_scanner_io(&ctx, tokio::fs::metadata(&item.path))
-                            .await?
-                            .map(|metadata| metadata.len())
-                            .unwrap_or(0)
+                        wait_for_scanner_io(
+                            &ctx,
+                            run_scanner_node_operation(self.node_admission.as_ref(), &ctx, tokio::fs::metadata(&item.path)),
+                        )
+                        .await??
+                        .map(|metadata| metadata.len())
+                        .unwrap_or(0)
                     }
                 } else {
                     0
                 };
                 wait_for_scanner_io(&ctx, self.sleeper.admit_disk_io(1, budgeted_bytes)).await?;
 
-                self.object_metadata_visits.increment(1);
-                let sz = match self
-                    .local_disk
-                    .get_size_with_tier_names(item.clone(), &self.tier_registry.names)
-                    .await
+                let sz = match run_scanner_node_operation(self.node_admission.as_ref(), &ctx, async {
+                    self.object_metadata_visits.increment(1);
+                    self.local_disk
+                        .get_size_with_tier_names(item.clone(), &self.tier_registry.names)
+                        .await
+                })
+                .await?
                 {
                     Ok(sz) => sz,
                     Err(e) => {
@@ -2548,7 +2573,7 @@ impl FolderScanner {
                     {
                         Ok(outcome) => outcome,
                         Err(e) => {
-                            if ctx.is_cancelled() {
+                            if ctx.is_cancelled() || matches!(&e, ScannerError::ResourceDeferred) {
                                 self.preserve_partial_child_progress(&folder_item.parent, &h, into, &dst)
                                     .await;
                                 return Err(e);
@@ -2794,19 +2819,23 @@ impl FolderScanner {
                 let mut seen_truncated_objects: HashSet<String> = HashSet::new();
 
                 loop {
-                    if agreed_closed && partial_closed && finished_closed {
+                    if child_ctx.is_cancelled() || (agreed_closed && partial_closed && finished_closed) {
                         break;
                     }
+                    #[cfg(test)]
+                    let defer_entries = self.defer_abandoned_entries_until_finished && !finished_closed;
+                    #[cfg(not(test))]
+                    let defer_entries = false;
 
                     select! {
-                        entry_name = agreed_rx.recv(), if !agreed_closed => {
+                        entry_name = agreed_rx.recv(), if !agreed_closed && !defer_entries => {
                             let Some(entry_name) = entry_name else {
                                 agreed_closed = true;
                                 continue;
                             };
                             (self.update_current_path)(&entry_name).await;
                         }
-                        entries = partial_rx.recv(), if !partial_closed => {
+                        entries = partial_rx.recv(), if !partial_closed && !defer_entries => {
                             let Some(entries) = entries else {
                                 partial_closed = true;
                                 continue;
@@ -2932,6 +2961,7 @@ impl FolderScanner {
                                 finished_closed = true;
                                 continue;
                             };
+                            finished_closed = true;
                             if disk_errors_are_only_missing_paths(&errs) {
                                 debug!(
                                     target: "rustfs::scanner::folder",
@@ -2953,7 +2983,9 @@ impl FolderScanner {
                                     "Scanner list_path_raw finished with disk errors"
                                 );
                             }
-                            child_ctx.cancel();
+                            // The producer awaited callback sends, not their
+                            // consumption. Drain the remaining candidate and
+                            // agreed frames before closing this child walk.
                         }
                         _ = child_ctx.cancelled() => {
                             break;
@@ -2971,7 +3003,10 @@ impl FolderScanner {
                     if into.compacted {
                         // In compacted mode child totals are accumulated directly into the parent entry.
                         let fut = Box::pin(self.scan_folder(ctx.clone(), folder_item.clone(), into));
-                        fut.await.map_err(|e| ScannerError::Other(e.to_string()))?;
+                        fut.await.map_err(|error| match error {
+                            ScannerError::ResourceDeferred => ScannerError::ResourceDeferred,
+                            other => ScannerError::Other(other.to_string()),
+                        })?;
                         self.send_update_for_entry(&this_hash, &folder.parent, into).await;
                         tokio::task::yield_now().await;
                     } else {
@@ -2990,7 +3025,7 @@ impl FolderScanner {
                             )
                             .await
                         {
-                            if ctx.is_cancelled() {
+                            if ctx.is_cancelled() || matches!(&e, ScannerError::ResourceDeferred) {
                                 self.preserve_partial_child_progress(&folder_item.parent, &h, into, &dst)
                                     .await;
                                 return Err(e);
@@ -3197,6 +3232,7 @@ pub(crate) async fn scan_data_folder_scoped(
         failed_object_paths_seen: HashSet::new(),
         resolved_failed_object_paths: HashSet::new(),
         sleeper,
+        node_admission: crate::workload_admission::scanner_node_admission(),
         disks,
         disks_quorum,
         updates,
@@ -3229,6 +3265,8 @@ pub(crate) async fn scan_data_folder_scoped(
         pending_size_reconciliation_truncated: false,
         #[cfg(test)]
         list_path_raw_options_observer: None,
+        #[cfg(test)]
+        defer_abandoned_entries_until_finished: false,
     };
 
     let now = FolderScanner::now_secs();
@@ -3319,7 +3357,7 @@ pub(crate) async fn scan_data_folder_scoped(
             }
         }
         Err(e) => {
-            if ctx.is_cancelled() {
+            if ctx.is_cancelled() || matches!(&e, ScannerError::ResourceDeferred) {
                 let root_hash = hash_path(&bucket_name);
                 let root_has_progress = data_usage_root_has_progress(&root);
                 let pending_heals_changed = scanner.pending_heals_changed;
@@ -3394,6 +3432,19 @@ pub(crate) async fn scan_data_folder_scoped(
                 close_disk_guard.close().await;
                 return Err(ScannerError::NamespaceNotFoundCache(Box::new(partial_cache)));
             }
+            if matches!(&e, ScannerError::ResourceDeferred) {
+                // Admission pressure is not disk failure or an empty census.
+                // Keep the previous tree when no new useful progress exists.
+                let mut partial_cache = scanner.old_cache;
+                if partial_cache.cache.is_empty() {
+                    partial_cache
+                        .cache
+                        .insert(hash_path(&bucket_name).0, DataUsageEntry::default());
+                }
+                partial_cache.info.snapshot_complete = false;
+                close_disk_guard.close().await;
+                return Err(ScannerError::PartialCache(Box::new(partial_cache)));
+            }
             close_disk_guard.close().await;
             // No useful information, return original cache
             Err(e)

crates/scanner/src/scanner_folder/tests.rs

@@ -30,7 +30,9 @@ use std::sync::Mutex;
 pub(super) mod checkpoint_fixture;
 pub(super) mod enumeration_restart;
 mod incremental_enumeration;
+mod listing_completion;
 mod metadata_visits;
+mod node_admission;
 
 /// Reset the process-global alert cooldown map; test-only.
 fn reset_alert_cooldowns() {
@@ -364,6 +366,7 @@ async fn build_test_scanner() -> (FolderScanner, std::path::PathBuf) {
         failed_object_paths_seen: HashSet::new(),
         resolved_failed_object_paths: HashSet::new(),
         sleeper: SCANNER_SLEEPER.clone(),
+        node_admission: None,
         disks: Vec::new(),
         disks_quorum: 0,
         updates: None,
@@ -398,6 +401,7 @@ async fn build_test_scanner() -> (FolderScanner, std::path::PathBuf) {
         pending_size_reconciliation_scopes: HashSet::new(),
         pending_size_reconciliation_truncated: false,
         list_path_raw_options_observer: None,
+        defer_abandoned_entries_until_finished: false,
     };
 
     (scanner, temp_dir)

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

@@ -0,0 +1,104 @@
+// Copyright 2026 RustFS Team
+// Licensed under the Apache License, Version 2.0.
+
+use super::*;
+
+#[tokio::test]
+#[serial]
+async fn abandoned_listing_drains_candidate_buffered_before_finished() {
+    let (mut scanner, root) = build_test_scanner().await;
+    let _guard = TestGuard {
+        temp_dir: Some(root.clone()),
+    };
+    let bucket = "bucket";
+    let object = "abandoned/object";
+    tokio::fs::create_dir_all(root.join(bucket)).await.expect("local bucket");
+    let mut disks = vec![scanner.local_disk.clone()];
+    let mut original_metadata = Vec::new();
+    for (index, version) in [(2, 0x1935), (3, 0x1935), (4, 0x1934)] {
+        let disk_root = root.join(format!("disk-{index}"));
+        let metadata = metadata_for_object_version(bucket, object, Some(Uuid::from_u128(version)));
+        write_test_object_metadata_bytes(&disk_root, bucket, object, &metadata).await;
+        original_metadata.push((disk_root.join(bucket).join(object).join(STORAGE_FORMAT_FILE), metadata));
+        let endpoint = Endpoint::try_from(disk_root.to_string_lossy().as_ref()).expect("listing fixture endpoint");
+        disks.push(
+            new_disk(
+                &endpoint,
+                &DiskOption {
+                    cleanup: false,
+                    health_check: false,
+                },
+            )
+            .await
+            .expect("listing fixture disk"),
+        );
+    }
+    for cache in [&mut scanner.old_cache, &mut scanner.new_cache, &mut scanner.update_cache] {
+        cache.info.name = bucket.to_owned();
+    }
+    scanner.old_cache.replace(
+        "bucket/abandoned/object",
+        bucket,
+        DataUsageEntry {
+            objects: 1,
+            ..Default::default()
+        },
+    );
+    scanner.heal_object_select = 1;
+    scanner.disks = disks;
+    scanner.disks_quorum = 2;
+    scanner.sleeper = DynamicSleeper::new(rustfs_config::ScannerSpeed::Fastest);
+    // Force the valid select ordering that receives terminal errors before the
+    // final candidate. One candidate frame fits in the bounded partial channel.
+    scanner.defer_abandoned_entries_until_finished = true;
+    let (_commands, mut heal_rx) = rustfs_heal_contracts::heal_channel::init_heal_channels().expect("isolated heal channel");
+    let (captured_tx, mut captured_rx) = mpsc::unbounded_channel();
+    let responder = tokio::spawn(async move {
+        while let Some(command) = heal_rx.recv().await {
+            let rustfs_heal_contracts::heal_channel::HealReceiptCommand { request, response_tx } = command;
+            captured_tx
+                .send((request.object_prefix.clone(), request.object_version_id.clone()))
+                .expect("candidate capture remains open");
+            let _ = response_tx.send(Ok(rustfs_heal_contracts::heal_channel::HealAdmissionReceipt {
+                result: HealAdmissionResult::Accepted,
+                task_id: request.id,
+            }));
+        }
+    });
+    let result = tokio::time::timeout(
+        Duration::from_secs(2),
+        scanner.scan_folder(
+            CancellationToken::new(),
+            CachedFolder {
+                name: bucket.to_owned(),
+                parent: None,
+                object_heal_prob_div: 1,
+            },
+            &mut DataUsageEntry::default(),
+        ),
+    )
+    .await;
+    responder.abort();
+    result
+        .expect("listing completion must close every callback channel")
+        .expect("complete folder scan");
+    let mut candidates = HashSet::new();
+    while let Ok((object, version)) = captured_rx.try_recv() {
+        if let (Some(object), Some(version)) = (object, version) {
+            candidates.insert((object, version));
+        }
+    }
+    let expected = [0x1934, 0x1935]
+        .into_iter()
+        // Raw base-object listings include the trailing separator. Preserve
+        // the producer's candidate identity when draining its callback frame.
+        .map(|version| (format!("{object}/"), Uuid::from_u128(version).to_string()))
+        .collect();
+    assert_eq!(
+        candidates, expected,
+        "terminal disk errors must not discard buffered exact-version candidates"
+    );
+    for (path, metadata) in original_metadata {
+        assert_eq!(tokio::fs::read(path).await.expect("retained metadata"), metadata);
+    }
+}

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

@@ -0,0 +1,547 @@
+// Copyright 2026 RustFS Team
+// Licensed under the Apache License, Version 2.0.
+
+use super::*;
+use rustfs_concurrency::adaptive::{FeedbackController, FeedbackObservation, NodeAdmission, ResourcePressure};
+use std::num::NonZeroU16;
+use std::sync::atomic::AtomicU64;
+
+fn owner(open: bool) -> Arc<NodeAdmission> {
+    let owner = NodeAdmission::new(NonZeroU16::new(1).expect("one node unit"));
+    if open {
+        owner.update(
+            &mut FeedbackController::new(),
+            Some(FeedbackObservation {
+                generation: 1,
+                max_concurrency: 1,
+                pressure: ResourcePressure::Unknown,
+                completed_units: None,
+            }),
+        );
+    }
+    owner
+}
+
+#[tokio::test]
+#[serial]
+async fn scanner_node_admission_blocks_real_walk_before_open_or_metadata() {
+    let (mut scanner, root) = build_test_scanner().await;
+    let _guard = TestGuard {
+        temp_dir: Some(root.clone()),
+    };
+    write_test_object_metadata(&root, "bucket", "object").await;
+    let visits = super::metadata_visits::observe(&mut scanner);
+    let gate = owner(false);
+    scanner.node_admission = Some(gate.clone());
+    scanner.sleeper = DynamicSleeper::new(rustfs_config::ScannerSpeed::Fastest);
+    for name in ["bucket", "does-not-exist"] {
+        let result = scanner
+            .scan_folder(
+                CancellationToken::new(),
+                CachedFolder {
+                    name: name.to_string(),
+                    parent: None,
+                    object_heal_prob_div: 1,
+                },
+                &mut DataUsageEntry::default(),
+            )
+            .await;
+        assert!(
+            matches!(result, Err(ScannerError::ResourceDeferred)),
+            "node gate must precede both real reads and missing-path lookup: {result:?}"
+        );
+    }
+    assert_eq!(visits.load(Ordering::Relaxed), 0);
+    assert_eq!(gate.snapshot().active, 0);
+    assert_eq!(gate.completed(), Some(0));
+}
+
+#[tokio::test]
+#[serial]
+async fn scanner_node_admission_shares_capacity_with_heal_owner() {
+    let (mut scanner, root) = build_test_scanner().await;
+    let _guard = TestGuard {
+        temp_dir: Some(root.clone()),
+    };
+    write_test_object_metadata(&root, "bucket", "object").await;
+    let gate = owner(true);
+    let heal_unit = gate.acquire().await;
+    scanner.node_admission = Some(gate.clone());
+    scanner.sleeper = DynamicSleeper::new(rustfs_config::ScannerSpeed::Fastest);
+    let folder = CachedFolder {
+        name: "bucket".to_string(),
+        parent: None,
+        object_heal_prob_div: 1,
+    };
+    let result = scanner
+        .scan_folder(CancellationToken::new(), folder.clone(), &mut DataUsageEntry::default())
+        .await;
+    assert!(
+        matches!(result, Err(ScannerError::ResourceDeferred)),
+        "scanner must not create a separate node budget: {result:?}"
+    );
+    assert_eq!(gate.snapshot().active, 1);
+    heal_unit.complete();
+    let mut entry = DataUsageEntry::default();
+    scanner
+        .scan_folder(CancellationToken::new(), folder, &mut entry)
+        .await
+        .expect("released heal capacity admits the real scan");
+    assert_eq!(
+        scanner
+            .new_cache
+            .size_recursive("bucket")
+            .expect("complete bucket tree")
+            .objects,
+        1
+    );
+    assert_eq!(gate.snapshot().active, 0);
+    assert!(
+        gate.completed().is_some_and(|count| count > 1),
+        "successful scanner work must reach shared feedback"
+    );
+}
+
+#[tokio::test(start_paused = true)]
+async fn scanner_node_admission_precancellation_never_polls_storage() {
+    let gate = owner(false);
+    let cancel = CancellationToken::new();
+    let polled = std::cell::Cell::new(false);
+    cancel.cancel();
+    let operation = run_scanner_node_operation(Some(&gate), &cancel, async {
+        polled.set(true);
+        Ok::<_, std::io::Error>(())
+    });
+    assert!(matches!(operation.await, Err(ScannerError::Other(message)) if message == "Operation cancelled"));
+    assert!(!polled.get());
+    assert_eq!(gate.completed(), Some(0));
+    assert_eq!(gate.snapshot().active, 0);
+}
+
+#[tokio::test(start_paused = true)]
+async fn scanner_node_admission_shrink_keeps_running_unit_and_preserves_errors() {
+    let gate = owner(true);
+    let cancel = CancellationToken::new();
+    let mut running = Box::pin(run_scanner_node_operation(
+        Some(&gate),
+        &cancel,
+        std::future::pending::<Result<(), std::io::Error>>(),
+    ));
+    assert!(futures::poll!(&mut running).is_pending());
+    assert_eq!(gate.snapshot().active, 1);
+    gate.update(&mut FeedbackController::new(), None);
+    let polled = std::cell::Cell::new(false);
+    let deferred = run_scanner_node_operation(Some(&gate), &cancel, async {
+        polled.set(true);
+        Ok::<_, std::io::Error>(())
+    })
+    .await;
+    assert!(matches!(deferred, Err(ScannerError::ResourceDeferred)));
+    assert!(!polled.get());
+    assert_eq!(gate.snapshot().active, 1, "shrink cannot revoke admitted work");
+    drop(running);
+    assert_eq!(gate.snapshot().active, 0);
+    assert_eq!(gate.completed(), Some(0), "abandoned work is not progress");
+    gate.update(
+        &mut FeedbackController::new(),
+        Some(FeedbackObservation {
+            generation: 2,
+            max_concurrency: 1,
+            pressure: ResourcePressure::Unknown,
+            completed_units: None,
+        }),
+    );
+    let error = run_scanner_node_operation(Some(&gate), &cancel, async {
+        Err::<(), _>(std::io::Error::from(std::io::ErrorKind::PermissionDenied))
+    })
+    .await
+    .expect("resource admission succeeds")
+    .expect_err("storage error survives");
+    assert_eq!(error.kind(), std::io::ErrorKind::PermissionDenied);
+    assert_eq!(gate.completed(), Some(0));
+    run_scanner_node_operation(Some(&gate), &cancel, async { Ok::<_, std::io::Error>(()) })
+        .await
+        .expect("admitted")
+        .expect("successful work");
+    assert_eq!(gate.completed(), Some(1));
+    assert_eq!(gate.snapshot().active, 0);
+}
+
+struct ShrinkOnMetadata {
+    owner: Arc<NodeAdmission>,
+    visited: AtomicU64,
+}
+impl metrics::CounterFn for ShrinkOnMetadata {
+    fn increment(&self, value: u64) {
+        if self.visited.fetch_add(value, Ordering::SeqCst) == 0 {
+            self.owner.update(&mut FeedbackController::new(), None);
+            assert_eq!(self.owner.snapshot().active, 1, "shrink occurs during the admitted object");
+        }
+    }
+    fn absolute(&self, _: u64) {
+        panic!("visit counter must be cumulative");
+    }
+}
+struct ShrinkRecorder(Arc<ShrinkOnMetadata>);
+impl metrics::Recorder for ShrinkRecorder {
+    fn describe_counter(&self, _: metrics::KeyName, _: Option<metrics::Unit>, _: metrics::SharedString) {}
+    fn describe_gauge(&self, _: metrics::KeyName, _: Option<metrics::Unit>, _: metrics::SharedString) {}
+    fn describe_histogram(&self, _: metrics::KeyName, _: Option<metrics::Unit>, _: metrics::SharedString) {}
+    fn register_counter(&self, key: &metrics::Key, _: &metrics::Metadata<'_>) -> metrics::Counter {
+        if key.name() == "rustfs_scanner_object_metadata_visits_total" {
+            metrics::Counter::from_arc(self.0.clone())
+        } else {
+            metrics::Counter::noop()
+        }
+    }
+    fn register_gauge(&self, _: &metrics::Key, _: &metrics::Metadata<'_>) -> metrics::Gauge {
+        metrics::Gauge::noop()
+    }
+    fn register_histogram(&self, _: &metrics::Key, _: &metrics::Metadata<'_>) -> metrics::Histogram {
+        metrics::Histogram::noop()
+    }
+}
+
+#[tokio::test]
+#[serial]
+async fn scanner_node_admission_mid_walk_shrink_preserves_partial_and_resumes_all_objects() {
+    let (scanner, root) = build_test_scanner().await;
+    let _guard = TestGuard {
+        temp_dir: Some(root.clone()),
+    };
+    for path in ["prefix/a/first", "prefix/b/second"] {
+        write_test_object_metadata(&root, "bucket", path).await;
+    }
+    let gate = owner(true);
+    let recorder = ShrinkRecorder(Arc::new(ShrinkOnMetadata {
+        owner: gate.clone(),
+        visited: AtomicU64::new(0),
+    }));
+    let scan = |cache| {
+        let cancel = CancellationToken::new();
+        scan_data_folder(
+            cancel.clone(),
+            ScannerCycleBudget::new(&cancel, Default::default()),
+            vec![scanner.local_disk.clone()],
+            scanner.local_disk.clone(),
+            cache,
+            None,
+            HealScanMode::Normal,
+            DynamicSleeper::new(rustfs_config::ScannerSpeed::Fastest),
+        )
+    };
+    let cache = DataUsageCache {
+        info: crate::DataUsageCacheInfo {
+            name: "bucket".to_string(),
+            skip_healing: true,
+            ..Default::default()
+        },
+        ..Default::default()
+    };
+    let baseline = scan(cache.clone()).await.expect("independent full walk");
+    let partial = crate::workload_admission::TEST_SCANNER_NODE_ADMISSION
+        .scope(gate.clone(), async {
+            let mut scanning = std::pin::pin!(scan(cache));
+            std::future::poll_fn(|cx| metrics::with_local_recorder(&recorder, || scanning.as_mut().poll(cx))).await
+        })
+        .await;
+    let Err(ScannerError::PartialCache(partial)) = partial else {
+        panic!("deferral below root cannot certify completion: {partial:?}");
+    };
+    assert!(!partial.info.snapshot_complete);
+    assert_eq!(
+        partial
+            .size_recursive("bucket")
+            .expect("partial keeps accounted work")
+            .objects,
+        1
+    );
+    assert_eq!(recorder.0.visited.load(Ordering::SeqCst), 1);
+    assert_eq!(gate.snapshot().active, 0);
+    gate.update(
+        &mut FeedbackController::new(),
+        Some(FeedbackObservation {
+            generation: 2,
+            max_concurrency: 1,
+            pressure: ResourcePressure::Unknown,
+            completed_units: None,
+        }),
+    );
+    let resumed = crate::workload_admission::TEST_SCANNER_NODE_ADMISSION
+        .scope(gate.clone(), scan(*partial))
+        .await
+        .expect("resume completes with restored node capacity");
+    assert!(resumed.info.snapshot_complete);
+    let expected = baseline.size_recursive("bucket").expect("baseline bucket");
+    let actual = resumed.size_recursive("bucket").expect("resumed bucket");
+    assert_eq!(
+        (actual.objects, actual.size, actual.versions),
+        (expected.objects, expected.size, expected.versions)
+    );
+    assert_eq!(actual.objects, 2);
+    assert_eq!(gate.snapshot().active, 0);
+}
+
+#[tokio::test(start_paused = true)]
+async fn scanner_node_admission_does_not_park_while_cache_lease_may_be_held() {
+    let gate = owner(false);
+    let cancel = CancellationToken::new();
+    let polled = std::cell::Cell::new(false);
+    let mut operation = Box::pin(run_scanner_node_operation(Some(&gate), &cancel, async {
+        polled.set(true);
+        Ok::<_, std::io::Error>(())
+    }));
+    assert!(matches!(
+        futures::poll!(&mut operation),
+        std::task::Poll::Ready(Err(ScannerError::ResourceDeferred))
+    ));
+    assert!(!polled.get());
+    assert_eq!(gate.snapshot().active, 0);
+}
+
+#[tokio::test]
+#[serial]
+async fn scanner_node_admission_early_deferral_keeps_old_tree_incomplete() {
+    let (scanner, root) = build_test_scanner().await;
+    let _guard = TestGuard { temp_dir: Some(root) };
+    let mut cache = DataUsageCache {
+        info: crate::DataUsageCacheInfo {
+            name: "bucket".to_string(),
+            skip_healing: true,
+            snapshot_complete: true,
+            ..Default::default()
+        },
+        ..Default::default()
+    };
+    cache.replace(
+        "bucket",
+        "",
+        DataUsageEntry {
+            objects: 7,
+            size: 19,
+            ..Default::default()
+        },
+    );
+    let gate = owner(false);
+    let cancel = CancellationToken::new();
+    let result = crate::workload_admission::TEST_SCANNER_NODE_ADMISSION
+        .scope(
+            gate.clone(),
+            scan_data_folder(
+                cancel.clone(),
+                ScannerCycleBudget::new(&cancel, Default::default()),
+                vec![scanner.local_disk.clone()],
+                scanner.local_disk,
+                cache,
+                None,
+                HealScanMode::Normal,
+                DynamicSleeper::new(rustfs_config::ScannerSpeed::Fastest),
+            ),
+        )
+        .await;
+    let Err(ScannerError::PartialCache(partial)) = result else {
+        panic!("resource pressure is a partial census, not a disk error: {result:?}");
+    };
+    assert!(!partial.info.snapshot_complete);
+    let totals = partial.size_recursive("bucket").expect("old tree retained");
+    assert_eq!((totals.objects, totals.size), (7, 19));
+    assert_eq!(gate.completed(), Some(0));
+}
+
+#[tokio::test]
+#[serial]
+async fn scanner_node_admission_compacted_heal_child_preserves_deferral() {
+    let (mut scanner, root) = build_test_scanner().await;
+    let _guard = TestGuard {
+        temp_dir: Some(root.clone()),
+    };
+    let bucket = "bucket";
+    let object = "prefix/abandoned";
+    for index in 0..DATA_SCANNER_COMPACT_AT_FOLDERS {
+        tokio::fs::create_dir_all(root.join(bucket).join("prefix").join(format!("child-{index:04}")))
+            .await
+            .expect("create real compaction threshold");
+    }
+    let mut disks = vec![scanner.local_disk.clone()];
+    for (index, version) in [(2, 0x1935), (3, 0x1935), (4, 0x1934)] {
+        let disk_root = root.join(format!("disk-{index}"));
+        // Two partial frames exceed the coordinator channel's capacity, so
+        // discovery must consume the first before the finished frame can race it.
+        for child in ["first", "second"] {
+            let child_object = format!("{object}/{child}");
+            let metadata = metadata_for_object_version(bucket, &child_object, Some(Uuid::from_u128(version)));
+            write_test_object_metadata_bytes(&disk_root, bucket, &child_object, &metadata).await;
+        }
+        let endpoint = Endpoint::try_from(disk_root.to_string_lossy().as_ref()).expect("remote fixture endpoint");
+        disks.push(
+            new_disk(
+                &endpoint,
+                &DiskOption {
+                    cleanup: false,
+                    health_check: false,
+                },
+            )
+            .await
+            .expect("remote fixture disk"),
+        );
+    }
+    scanner.old_cache.info.name = bucket.to_string();
+    scanner.new_cache.info.name = bucket.to_string();
+    scanner.update_cache.info.name = bucket.to_string();
+    scanner.old_cache.replace(
+        &format!("{bucket}/{object}"),
+        "bucket/prefix",
+        DataUsageEntry {
+            objects: 1,
+            ..Default::default()
+        },
+    );
+    scanner.heal_object_select = 1;
+    scanner.disks = disks;
+    scanner.disks_quorum = 2;
+    scanner.sleeper = DynamicSleeper::new(rustfs_config::ScannerSpeed::Fastest);
+    let gate = owner(true);
+    scanner.node_admission = Some(gate.clone());
+    let shrunk = Arc::new(AtomicBool::new(false));
+    let response_gate = gate.clone();
+    let response_shrunk = shrunk.clone();
+    let (_commands, mut heal_rx) = rustfs_heal_contracts::heal_channel::init_heal_channels().expect("isolated heal channel");
+    let responder = tokio::spawn(async move {
+        while let Some(command) = heal_rx.recv().await {
+            let rustfs_heal_contracts::heal_channel::HealReceiptCommand { request, response_tx } = command;
+            if request.object_version_id.is_some() {
+                response_shrunk.store(true, Ordering::SeqCst);
+                response_gate.update(&mut FeedbackController::new(), None);
+            }
+            let _ = response_tx.send(Ok(rustfs_heal_contracts::heal_channel::HealAdmissionReceipt {
+                result: HealAdmissionResult::Accepted,
+                task_id: request.id,
+            }));
+        }
+    });
+    let mut into = DataUsageEntry::default();
+    let folder = CachedFolder {
+        name: "bucket/prefix".to_string(),
+        parent: Some(hash_path(bucket)),
+        object_heal_prob_div: 1,
+    };
+    let result = {
+        let mut walk = std::pin::pin!(scanner.scan_folder(CancellationToken::new(), folder, &mut into));
+        let mut controller = FeedbackController::new();
+        let mut generation = 0;
+        std::future::poll_fn(|cx| {
+            // Keep telemetry fresh until the real heal-discovery receipt shrinks it.
+            if !shrunk.load(Ordering::SeqCst) {
+                generation += 1;
+                gate.update(
+                    &mut controller,
+                    Some(FeedbackObservation {
+                        generation,
+                        max_concurrency: 1,
+                        pressure: ResourcePressure::Unknown,
+                        completed_units: None,
+                    }),
+                );
+            }
+            walk.as_mut().poll(cx)
+        })
+        .await
+    };
+    responder.abort();
+    assert!(into.compacted, "real directory count must enter the compacted recursion branch");
+    assert!(shrunk.load(Ordering::SeqCst), "remote orphan discovery must submit an exact-version heal");
+    assert!(
+        matches!(result, Err(ScannerError::ResourceDeferred)),
+        "compacted recursion must retain typed resource pressure for partial-cache handling: {result:?}"
+    );
+    assert_eq!(gate.snapshot().active, 0);
+}
+
+#[tokio::test]
+#[serial]
+async fn scanner_node_admission_progresses_during_continuous_heal_queue() {
+    let (mut scanner, root) = build_test_scanner().await;
+    let _guard = TestGuard {
+        temp_dir: Some(root.clone()),
+    };
+    write_test_object_metadata(&root, "bucket", "object").await;
+    for index in 0..65 {
+        tokio::fs::write(root.join("bucket/object").join(format!("part.{index}.integrity.orphan")), b"")
+            .await
+            .expect("sidecars require more than one directory refill");
+    }
+    scanner.is_erasure_mode = true;
+    scanner.budget = ScannerCycleBudget::new_with_progress_tracking(&CancellationToken::new(), Default::default());
+    let visits = super::metadata_visits::observe(&mut scanner);
+    let gate = owner(true);
+    scanner.node_admission = Some(gate.clone());
+    scanner.sleeper = DynamicSleeper::new(rustfs_config::ScannerSpeed::Fastest);
+    let heal_stop = CancellationToken::new();
+    let heal_stop_worker = heal_stop.clone();
+    let heal_gate = gate.clone();
+    let heal_completions = Arc::new(AtomicU64::new(0));
+    let completed = heal_completions.clone();
+    let (started, ready) = tokio::sync::oneshot::channel();
+    let heals = tokio::spawn(async move {
+        let mut held = heal_gate.acquire().await;
+        started.send(()).expect("scanner waits for the first heal");
+        loop {
+            let next = heal_gate.acquire();
+            tokio::pin!(next);
+            assert!(futures::poll!(&mut next).is_pending());
+            held.complete();
+            completed.fetch_add(1, Ordering::Relaxed);
+            held = tokio::select! {
+                biased;
+                _ = heal_stop_worker.cancelled() => return,
+                permit = &mut next => permit,
+            };
+            tokio::task::yield_now().await;
+        }
+    });
+    ready.await.expect("heal starts before scanner");
+    let (node_tx, mut node_rx) = mpsc::channel::<crate::workload_admission::ScannerNodeRequest>(1);
+    let folder = CachedFolder {
+        name: "bucket".to_string(),
+        parent: None,
+        object_heal_prob_div: 1,
+    };
+    let mut handoffs = 0;
+    let result = tokio::time::timeout(Duration::from_secs(2), async {
+        let mut totals = DataUsageEntry::default();
+        let scan = crate::workload_admission::with_scanner_node_requests(
+            Some(node_tx),
+            scanner.scan_folder(CancellationToken::new(), folder, &mut totals),
+        );
+        tokio::pin!(scan);
+        loop {
+            tokio::select! {
+                result = &mut scan => break result,
+                Some(request) = node_rx.recv() => {
+                    handoffs += 1;
+                    // This direct folder fixture owns no cache lock. Production
+                    // coordinators release and revalidate their lease here.
+                    let permit = request.owner.acquire().await;
+                    assert_eq!(gate.snapshot().active, 1);
+                    request.respond.send(permit).unwrap_or_else(|_| panic!("real walk still waits"));
+                }
+            }
+        }
+    })
+    .await;
+    heal_stop.cancel();
+    heals.await.expect("continuous heal worker exits");
+    result
+        .expect("scanner must finish while Heal keeps queueing")
+        .expect("real directory scan");
+    assert_eq!(scanner.new_cache.size_recursive("bucket").expect("complete tree").objects, 1);
+    assert!(handoffs > 1, "the walk must yield through multiple shared-node turns");
+    assert_eq!(scanner.budget.entries_visited(), 67, "resumption must not replay directory entries");
+    assert_eq!(
+        visits.load(Ordering::Relaxed),
+        1,
+        "classification must account for the object exactly once"
+    );
+    assert!(heal_completions.load(Ordering::Relaxed) > 0, "Heal must also advance during the walk");
+    assert_eq!(gate.snapshot().active, 0);
+}

crates/scanner/src/scanner_io/cache.rs

@@ -172,12 +172,55 @@ fn record_checkpoint_save_failure(cache_name: &str, checkpoint: &DataUsageCache,
 
 #[derive(Debug)]
 pub(crate) struct ScannerCacheLockGuards {
-    scoped: NamespaceLockGuard,
+    scoped: Option<NamespaceLockGuard>,
 }
 
 impl ScannerCacheLockGuards {
     pub(crate) fn is_lock_lost(&self) -> bool {
-        self.scoped.is_lock_lost()
+        self.scoped.as_ref().is_none_or(NamespaceLockGuard::is_lock_lost)
+    }
+
+    /// Suspend the same traversal while yielding its cache lease. Lock order on
+    /// resumption is node admission -> cache lease -> cache revision reads.
+    pub(crate) async fn resume_node_operation(
+        &mut self,
+        store: Arc<SetDisks>,
+        cache_name: &str,
+        source: DataUsageCacheSource,
+        revisions: &DataUsageCacheRevisions,
+        request: crate::workload_admission::ScannerNodeRequest,
+    ) -> Result<()> {
+        if self.is_lock_lost() {
+            return Err(Error::other("scanner cache lock was lost before node admission"));
+        }
+        if request.respond.is_closed() {
+            return Ok(());
+        }
+        let crate::workload_admission::ScannerNodeRequest { owner, respond, cancel } = request;
+        drop(self.scoped.take());
+        let resume = async {
+            let permit = owner.acquire().await;
+            *self = acquire_scanner_cache_locks(store.as_ref(), cache_name, source)
+                .await
+                .map_err(Error::other)?;
+            let current = crate::storage_api::ecstore_with_background_disk_io(
+                cancel.clone(),
+                DataUsageCache::read_revisions(store, cache_name),
+            )
+            .await?;
+            if self.is_lock_lost() || &current != revisions {
+                return Err(Error::other("scanner cache changed while waiting for node admission"));
+            }
+            // If the walk was dropped, send returns ownership and drops the
+            // permit without recording completed work.
+            let _ = respond.send(permit);
+            Ok(())
+        };
+        tokio::select! {
+            biased;
+            _ = cancel.cancelled() => Err(Error::other("scanner node admission cancelled")),
+            result = resume => result,
+        }
     }
 }
 
@@ -215,6 +258,15 @@ impl std::fmt::Display for ScannerCacheLockError {
     }
 }
 
+impl std::error::Error for ScannerCacheLockError {
+    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
+        match self {
+            Self::Create { source, .. } => Some(source),
+            Self::Acquire { source, .. } => Some(source),
+        }
+    }
+}
+
 pub(crate) async fn acquire_scanner_cache_locks(
     store: &SetDisks,
     cache_name: &str,
@@ -237,7 +289,7 @@ pub(crate) async fn acquire_scanner_cache_locks(
             source,
         })?;
 
-    Ok(ScannerCacheLockGuards { scoped })
+    Ok(ScannerCacheLockGuards { scoped: Some(scoped) })
 }
 
 pub(super) async fn await_scanner_disk_shutdown<F>(scan: Pin<&mut F>)

crates/scanner/src/scanner_io/io_cache.rs

@@ -1064,7 +1064,7 @@ impl ScannerIOCache for SetDisks {
 
                     // Lock order: scanner leader fence -> set-scoped per-bucket cache lock ->
                     // cache object read/write.
-                    let cache_guard = match acquire_scanner_cache_locks(store_clone_clone.as_ref(), &cache_name, source).await {
+                    let mut cache_guard = match acquire_scanner_cache_locks(store_clone_clone.as_ref(), &cache_name, source).await {
                         Ok(guard) => guard,
                         Err(e) => {
                             if e.is_contention() {
@@ -1294,9 +1294,10 @@ impl ScannerIOCache for SetDisks {
                     );
 
                     let (checkpoint_tx, mut checkpoint_rx) = mpsc::channel::<DataUsageCache>(1);
+                    let (node_tx, mut node_rx) = crate::workload_admission::scanner_node_request_channel();
 
                     let scan_ctx = ctx_clone.child_token();
-                    let scan = disk.clone().nsscanner_disk(
+                    let scan = crate::workload_admission::with_scanner_node_requests(node_tx, disk.clone().nsscanner_disk(
                         scan_ctx.clone(),
                         budget_clone.clone(),
                         set_disk_inventory_clone.as_ref().clone(),
@@ -1307,7 +1308,7 @@ impl ScannerIOCache for SetDisks {
                             prefix_scan_scope,
                             checkpoint_tx: Some(checkpoint_tx),
                         },
-                    );
+                    ));
                     tokio::pin!(scan);
                     let mut lock_watch = tokio::time::interval(SCANNER_CACHE_LOCK_POLL_INTERVAL);
                     lock_watch.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
@@ -1315,6 +1316,24 @@ impl ScannerIOCache for SetDisks {
                     let scan_result = loop {
                         tokio::select! {
                             result = &mut scan => break result,
+                            request = async {
+                                match node_rx.as_mut() {
+                                    Some(receiver) => receiver.recv().await,
+                                    None => std::future::pending().await,
+                                }
+                            } => {
+                                let Some(request) = request else {
+                                    node_rx = None;
+                                    continue;
+                                };
+                                if let Err(error) = cache_guard.resume_node_operation(
+                                    store_clone_clone.clone(), &cache_name, source, &revisions, request,
+                                ).await {
+                                    scan_ctx.cancel();
+                                    await_scanner_disk_shutdown(scan.as_mut()).await;
+                                    break Err(error);
+                                }
+                            }
                             _ = lock_watch.tick() => {
                                 if cache_guard.is_lock_lost() {
                                     scan_ctx.cancel();

crates/scanner/src/scanner_io/tests.rs

@@ -47,6 +47,7 @@ use uuid::Uuid;
 mod checkpoint_coverage_identity;
 mod collector_lifecycle;
 mod listing_window_scope;
+mod node_admission;
 mod partial_bucket_carry;
 mod per_bucket_certification;
 mod scoped_entry_fallback;

crates/scanner/src/scanner_io/tests/node_admission.rs

@@ -0,0 +1,136 @@
+// Copyright 2026 RustFS Team
+// Licensed under the Apache License, Version 2.0.
+
+use super::*;
+use crate::workload_admission::ScannerNodeRequest;
+use rustfs_concurrency::adaptive::{FeedbackController, FeedbackObservation, NodeAdmission, ResourcePressure};
+use std::num::NonZeroU16;
+use tokio::sync::oneshot;
+
+#[derive(Clone, Copy, PartialEq, Eq)]
+enum ResumeCase {
+    Unchanged,
+    Updated,
+    Cancelled,
+}
+
+async fn check_cache_lease_resume(case: ResumeCase) {
+    let (_directory, store) = setup_local_scanner_store(1, 4).await;
+    let set = store.pools[0].disk_set[0].clone();
+    let source = DataUsageCacheSource::new(0, 0);
+    let cache_name = "node-resume/.usage-cache.bin";
+    let cache = DataUsageCache::default();
+    cache.save(set.clone(), cache_name).await.expect("seed cache");
+    let revisions = DataUsageCache::read_revisions(set.clone(), cache_name)
+        .await
+        .expect("initial revisions");
+    let gate = NodeAdmission::new(NonZeroU16::new(1).expect("one shared unit"));
+    gate.update(
+        &mut FeedbackController::new(),
+        Some(FeedbackObservation {
+            generation: 1,
+            max_concurrency: 1,
+            pressure: ResourcePressure::Unknown,
+            completed_units: None,
+        }),
+    );
+    let mut held = Some(gate.acquire().await);
+    let mut guard = acquire_scanner_cache_locks(set.as_ref(), cache_name, source)
+        .await
+        .expect("initial cache lease");
+    let ctx = CancellationToken::new();
+    let (respond, response) = oneshot::channel();
+    let request = ScannerNodeRequest {
+        owner: gate.clone(),
+        respond,
+        cancel: ctx.clone(),
+    };
+    let mut expected = revisions.clone();
+    let result = {
+        let resume = guard.resume_node_operation(set.clone(), cache_name, source, &revisions, request);
+        tokio::pin!(resume);
+        assert!(futures::poll!(&mut resume).is_pending(), "shared node is occupied");
+        let other = tokio::time::timeout(Duration::from_secs(1), acquire_scanner_cache_locks(set.as_ref(), cache_name, source))
+            .await
+            .expect("cache lock must be released before joining the node queue")
+            .expect("another cache worker can acquire the lease");
+        if case == ResumeCase::Updated {
+            let mut newer = cache;
+            newer.info.next_cycle = 1;
+            newer
+                .save(set.clone(), cache_name)
+                .await
+                .expect("another worker updates the cache");
+            expected = DataUsageCache::read_revisions(set.clone(), cache_name)
+                .await
+                .expect("new cache revisions");
+            assert_ne!(expected, revisions);
+        }
+        drop(other);
+        if case == ResumeCase::Cancelled {
+            ctx.cancel();
+        } else {
+            held.take().expect("held node unit").complete();
+        }
+        tokio::time::timeout(Duration::from_secs(2), resume)
+            .await
+            .expect("node resumption must finish or observe cancellation")
+    };
+    if case == ResumeCase::Unchanged {
+        result.expect("unchanged cache resumes the same walk");
+        let permit = response.await.expect("coordinator returns the node permit");
+        assert_eq!(gate.snapshot().active, 1);
+        assert_eq!(gate.completed(), Some(1), "admitting scanner work must not count it as completed");
+        assert!(!guard.is_lock_lost());
+        let lock = set
+            .new_ns_lock(RUSTFS_META_BUCKET, &scanner_cache_lock_resource(cache_name, source))
+            .await
+            .expect("competing cache lock");
+        assert!(
+            lock.get_write_lock_quiet(Duration::from_millis(20)).await.is_err(),
+            "cache lease must be restored before the operation resumes"
+        );
+        permit.complete();
+    } else {
+        result.expect_err("cancelled or stale work cannot resume");
+        assert!(response.await.is_err(), "failed resumption must not authorize storage work");
+        if case == ResumeCase::Cancelled {
+            assert!(guard.is_lock_lost());
+            assert_eq!(gate.snapshot().active, 1, "the running heal keeps its unit");
+            drop(held.take());
+            assert_eq!(gate.completed(), Some(0));
+        } else {
+            assert_eq!(gate.completed(), Some(1));
+        }
+    }
+    drop(guard);
+    assert_eq!(gate.snapshot().active, 0);
+    assert!(gate.try_acquire().is_some(), "dropped requests must leave no queue or capacity leak");
+    assert_eq!(
+        DataUsageCache::read_revisions(set.clone(), cache_name)
+            .await
+            .expect("final revisions"),
+        expected
+    );
+    acquire_scanner_cache_locks(set.as_ref(), cache_name, source)
+        .await
+        .expect("final cache lease is available");
+}
+
+#[tokio::test]
+#[serial]
+async fn scanner_node_wait_yields_cache_lease_before_fifo_admission() {
+    check_cache_lease_resume(ResumeCase::Unchanged).await;
+}
+
+#[tokio::test]
+#[serial]
+async fn scanner_node_wait_rejects_cache_changed_while_unlocked() {
+    check_cache_lease_resume(ResumeCase::Updated).await;
+}
+
+#[tokio::test]
+#[serial]
+async fn scanner_node_wait_cancellation_releases_queue_without_stealing_running_unit() {
+    check_cache_lease_resume(ResumeCase::Cancelled).await;
+}

crates/scanner/src/workload_admission.rs

@@ -12,9 +12,113 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::sync::{Arc, LazyLock, RwLock};
+use std::sync::{Arc, LazyLock, OnceLock, RwLock};
 
+use rustfs_concurrency::adaptive::{NodeAdmission, NodePermit};
 use rustfs_concurrency::{AdmissionState, WorkloadAdmissionRegistrySnapshot, WorkloadAdmissionSnapshotProvider, WorkloadClass};
+use tokio::sync::{mpsc, oneshot};
+use tokio_util::sync::CancellationToken;
+
+use crate::ScannerError;
+
+static SCANNER_NODE_ADMISSION: OnceLock<Arc<NodeAdmission>> = OnceLock::new();
+
+tokio::task_local! {
+    static SCANNER_NODE_REQUESTS: ScannerNodeRequestSender;
+}
+
+pub(crate) struct ScannerNodeRequest {
+    pub(crate) owner: Arc<NodeAdmission>,
+    pub(crate) respond: oneshot::Sender<NodePermit>,
+    pub(crate) cancel: CancellationToken,
+}
+
+pub(crate) type ScannerNodeRequestSender = mpsc::Sender<ScannerNodeRequest>;
+pub(crate) type ScannerNodeRequestReceiver = mpsc::Receiver<ScannerNodeRequest>;
+
+/// One suspended operation per walk; disabled admission allocates no channel.
+pub(crate) fn scanner_node_request_channel() -> (Option<ScannerNodeRequestSender>, Option<ScannerNodeRequestReceiver>) {
+    if scanner_node_admission().is_none() {
+        return (None, None);
+    }
+    let (sender, receiver) = mpsc::channel(1);
+    (Some(sender), Some(receiver))
+}
+
+pub(crate) async fn with_scanner_node_requests<F: std::future::Future>(
+    sender: Option<ScannerNodeRequestSender>,
+    operation: F,
+) -> F::Output {
+    match sender {
+        Some(sender) => SCANNER_NODE_REQUESTS.scope(sender, operation).await,
+        None => operation.await,
+    }
+}
+
+#[cfg(test)]
+tokio::task_local! {
+    pub(crate) static TEST_SCANNER_NODE_ADMISSION: Arc<NodeAdmission>;
+}
+
+/// Install the heal manager's node owner before starting scanner workers.
+/// Refuse replacement: existing scans must not outlive an independently renewed owner.
+pub fn set_scanner_node_admission(owner: Arc<NodeAdmission>) -> bool {
+    SCANNER_NODE_ADMISSION.set(owner).is_ok()
+}
+
+pub(crate) fn scanner_node_admission() -> Option<Arc<NodeAdmission>> {
+    #[cfg(test)]
+    if let Ok(owner) = TEST_SCANNER_NODE_ADMISSION.try_with(Arc::clone) {
+        return Some(owner);
+    }
+    SCANNER_NODE_ADMISSION.get().cloned()
+}
+
+/// Try the fast path first. A scoped coordinator releases the cache lease before
+/// joining the node queue and validates the cache before returning a permit.
+/// Uncoordinated callers defer instead of parking under an unknown lock.
+pub(crate) async fn run_scanner_node_operation<T, E>(
+    owner: Option<&Arc<NodeAdmission>>,
+    cancel: &CancellationToken,
+    operation: impl std::future::Future<Output = Result<T, E>>,
+) -> Result<Result<T, E>, ScannerError> {
+    let Some(owner) = owner else {
+        return Ok(operation.await);
+    };
+    if cancel.is_cancelled() {
+        return Err(ScannerError::Other("Operation cancelled".to_string()));
+    }
+    let permit = match owner.try_acquire() {
+        Some(permit) => permit,
+        None => {
+            let sender = SCANNER_NODE_REQUESTS
+                .try_with(Clone::clone)
+                .map_err(|_| ScannerError::ResourceDeferred)?;
+            let acquire = async {
+                let (respond, response) = oneshot::channel();
+                sender
+                    .send(ScannerNodeRequest {
+                        owner: owner.clone(),
+                        respond,
+                        cancel: cancel.clone(),
+                    })
+                    .await
+                    .map_err(|_| ScannerError::ResourceDeferred)?;
+                response.await.map_err(|_| ScannerError::ResourceDeferred)
+            };
+            tokio::select! {
+                biased;
+                _ = cancel.cancelled() => return Err(ScannerError::Other("Operation cancelled".to_string())),
+                permit = acquire => permit?,
+            }
+        }
+    };
+    let result = operation.await;
+    if result.is_ok() {
+        permit.complete();
+    }
+    Ok(result)
+}
 
 type WorkloadSnapshotProviderRef = Arc<dyn WorkloadAdmissionSnapshotProvider + Send + Sync>;
 

docs/operations/scanner-runtime-controls.md

@@ -486,17 +486,21 @@ The OTLP exporter records these node-local histograms for completed manager admi
 
 The two new histograms aggregate all observed admission sources and results without per-request labels. An ordinary uninterrupted admission partitions the phase into wait and hold. Admin forceStart releases these guards while cancelling overlaps, then reacquires them: the cancellation gap is included only in the original total phase, and the second acquisition contributes to wait. Cancellation subroutine work is outside these outer admission guard intervals. Recording happens after all three guards are released. They measure manager admission ownership, not namespace/object-lock waiting during repair. The status field `maxLockPhaseMicros` remains a manager-lifetime maximum, not a p95 sample. Keep same-window histogram counts and bucket resolution when deriving percentiles; missing observations do not mean zero latency. These metrics do not establish performance acceptance by themselves.
 
-## Experimental heal node admission
+## Experimental shared node admission
 
-`RUSTFS_HEAL_ADAPTIVE_ADMISSION=true` enables Linux cgroup-v2 feedback for the four object-heal entry points owned by the heal manager. It is off by default while full SSD performance acceptance is pending. Unsupported platforms reject an enabled setting. This does not yet control scanner walks, bucket/format repair or physical devices, and it is not a replacement for existing worker, disk or byte budgets.
+`RUSTFS_HEAL_ADAPTIVE_ADMISSION=true` enables Linux cgroup-v2 feedback shared by the heal manager’s four object-heal entry points and local Scanner namespace reads and object accounting. It is off by default while full SSD performance acceptance is pending. Unsupported platforms reject an enabled setting. This does not yet control the abandoned-child streaming list coordinator, bucket/format repair or physical devices, and it is not a replacement for existing worker, disk or byte budgets.
 
 The node samples its effective CPU quota/cpuset, ancestor memory headroom, CPU usage/throttling and CPU/memory/I/O pressure once per second. Sampling uses one blocking worker at a time. Resource reads retain their real start/end times: reads spanning over 20 ms or older than one second after all inputs return cannot renew admission. Counter rates use both read brackets conservatively. The concurrency cap is at most the rounded-up effective CPU count and 256 scheduling units; existing operator limits still apply independently. At most 1% memory headroom blocks new work. This is a pressure safeguard, not a memory reservation or a claim that concurrent repair buffers fit in the remaining bytes.
 
 Each complete sample leases admission for three seconds. Missing resource caps block new work; missing or low-confidence feedback retains at most one maintenance unit. CPU use above 90%, throttling or pressure above 1% of elapsed time, memory headroom at most 10%, foreground errors/cancellations, a foreground queue or a comparable foreground p99 regression reduce concurrency immediately. Recovery requires CPU use at most 70%, throttling/stalls at most 0.1% of elapsed time, at least 20% memory headroom, fresh foreground evidence and actual successful work. The existing controller probes one additional unit only after eight consecutive healthy seconds and cooldown. Limits affect new admission and do not revoke active work.
 
+The Scanner captures the same node owner once per bucket walk. Directory open/refill, entry type and size lookup, and object accounting try admission before their first storage poll; each unit ends when its operation future finishes. On contention, local and remote coordinators suspend the same traversal, release the bucket cache lease, and join the shared FIFO queue. They reacquire the cache lease and verify both cache revisions before resuming. A changed cache rejects the suspended result. Cancellation also drops the queued request; the existing cycle budget still bounds the walk. This retains directory and recursion state while allowing Heal to advance at capacity one, without repeatedly replaying a partially classified directory.
+
+Calls without a coordinating cache lease retain immediate resource deferral and useful partial-cache handling. Existing remote single-session ownership and local-disk admission shares remain in force. Disabled admission creates no request channel. Enabled walks use one bounded channel, with a response allocation and cache revision reads only on contention; the uncontended per-entry path adds no global lookup or task. The streaming abandoned-child coordinator requires a separate ownership boundary: retaining a node unit while it waits for a bounded-channel consumer that needs the same unit would deadlock at capacity one.
+
 Foreground comparison uses the existing complete response-body windows and shared admission snapshots. Each active class needs at least 100 completions and a finite p99 interval; this permits the percentile rank but is not a statistical confidence guarantee. Comparisons require the same active classes and rates within 80–125% of a retained reference. A changed workload resets recovery. The reference is an observed workload window, not an independent background-off baseline. A p99 interval can justify stability only when its upper bound is within 5% of the reference lower bound; a lower bound above 105% of the reference upper bound is pressure. Overlapping/uncertain comparisons cannot justify growth. Missing queues, stale or repeated cuts, uncertain cut boundaries and low counts are not idle evidence.
 
-Admission waits at most 250 ms before returning the existing typed transient deferral, before polling storage or taking namespace locks. No-lock callers never wait. Queued read repair uses the same node gate; direct foreground ECStore calls remain outside this heal-manager adapter. Cancellation releases the non-cloneable node scheduling unit. Native work retains separate physical-I/O units, and unsuccessful/cancelled/dry-run work cannot increment the node's completed-work counter. This scheduling counter is not an authoritative heal receipt and cannot discharge repair responsibility.
+Heal admission waits at most 250 ms before returning the existing typed transient deferral, before polling storage or taking namespace locks. No-lock callers never wait. Queued read repair uses the same node gate; direct foreground ECStore calls remain outside this heal-manager adapter. Cancellation releases the non-cloneable node scheduling unit. Native work retains separate physical-I/O units, and unsuccessful/cancelled/dry-run work cannot increment the node's completed-work counter. This scheduling counter is not an authoritative heal receipt and cannot discharge repair responsibility.
 
 The `rustfs_background_node_admission_limit` and `rustfs_background_node_admission_active` gauges describe the last published sample, not a live reservation proof. A blocked sampler can leave these exports stale even though the admission lease has expired. Full disk-owner feedback, shared scanner scheduling, escaping-buffer byte ownership and the matched-release SSD matrix/hot-shrink/72-hour soak remain required before enabling this by default or claiming optimal performance.
 

rustfs/src/startup_background.rs

@@ -81,6 +81,14 @@ pub(crate) async fn init_background_service_runtime(store: Arc<ECStore>) -> Resu
         let admission = None;
         let heal_storage = Arc::new(ECStoreHealStorage::new(store.clone()).with_admission(admission.clone()));
         init_heal_manager_with_workload_provider(heal_storage, None, Some(workload_provider.clone())).await?;
+        if let Some(admission) = &admission
+            && !rustfs_scanner::set_scanner_node_admission(admission.clone())
+        {
+            return Err(std::io::Error::new(
+                std::io::ErrorKind::AlreadyExists,
+                "scanner node admission is already initialized",
+            ));
+        }
         #[cfg(any(target_os = "linux", all(test, unix)))]
         if let Some(admission) = admission {
             crate::background_admission::start(admission, background_cancel.clone(), workload_provider);

scripts/scanner_process_receipt.py

@@ -18,6 +18,7 @@ SAFE_ENVIRONMENT = frozenset({
     "RUSTFS_STORAGE_CLASS_STANDARD", "RUSTFS_DURABILITY_MODE", "RUSTFS_NEW_BUCKET_DURABILITY_MODE",
     "RUSTFS_SCANNER_ENABLED", "RUSTFS_HEAL_ENABLED", "RUSTFS_SCANNER_SPEED", "RUSTFS_SCANNER_IDLE_MODE",
     "RUSTFS_SCANNER_START_DELAY_SECS", "RUSTFS_SCANNER_CYCLE",
+    "RUSTFS_HEAL_ADAPTIVE_ADMISSION",
     "RUSTFS_OBJECT_LOCK_DIAG_ENABLE", "RUSTFS_OBS_LOGS_EXPORT_ENABLED", "RUSTFS_OBS_TRACES_EXPORT_ENABLED",
     "RUSTFS_OBS_METRICS_EXPORT_ENABLED", "RUST_LOG", "NO_PROXY", "no_proxy",
 })

scripts/test_scanner_process_receipt.py

@@ -108,6 +108,21 @@ class ProcessReceiptTest(unittest.TestCase):
         with self.assertRaisesRegex(ValueError, "not active"):
             self.attest()
 
+    def test_adaptive_admission_is_bound_to_actual_process_environment(self):
+        key = "RUSTFS_HEAL_ADAPTIVE_ADMISSION"
+        self.public[key] = "true"
+        raw = self.environment + b"RUSTFS_HEAL_ADAPTIVE_ADMISSION=true\0"
+        (self.pid / "environ").write_bytes(raw)
+        result = self.attest()
+        self.assertEqual(result["environment_sha256"], receipt.fingerprint(self.public))
+        self.assertIn(key, result["environment_keys"])
+        for observed in (self.environment, raw.replace(b"ADMISSION=true", b"ADMISSION=false"),
+                         raw + b"RUSTFS_HEAL_ADAPTIVE_ADMISSION=true\0"):
+            with self.subTest(raw_hash=hashlib.sha256(observed).hexdigest()), self.assertRaises(ValueError):
+                receipt.environment_digest(observed, self.public)
+        with self.assertRaisesRegex(ValueError, "mismatch"):
+            receipt.environment_digest(raw, {**self.public, key: "false"})
+
     def test_wrong_binary_content_is_rejected(self):
         self.binary.write_bytes(b"replaced executable")
         with self.assertRaisesRegex(ValueError, "binary digest"):