fix(scanner): correct expiry test result type

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/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs

@@ -1086,8 +1086,9 @@ impl ExpiryState {
         }))
     }
 
-    #[cfg(test)]
-    fn new_with_unconsumed_worker_channel(capacity: usize) -> Arc<RwLock<Self>> {
+    #[cfg(any(test, feature = "test-util"))]
+    #[doc(hidden)]
+    pub fn new_with_unconsumed_worker_channel(capacity: usize) -> Arc<RwLock<Self>> {
         let (tx, rx) = mpsc::channel(capacity);
         Arc::new(RwLock::new(Self {
             tasks_tx: vec![tx],

crates/ecstore/src/disk/local.rs

@@ -8450,6 +8450,23 @@ impl LocalDisk {
                     return Err(std::io::Error::other("metadata read worker cancelled"));
                 }
                 let _io_permit = io_permit;
+                #[cfg(all(any(test, feature = "test-util"), not(windows)))]
+                {
+                    let opened = root.open_readonly(&path);
+                    if opened.is_ok() {
+                        os::prepared_publication_test_hooks::run(
+                            os::prepared_publication_test_hooks::Stage::MetadataReadAfterOpen,
+                            &path,
+                        );
+                    }
+                    let result = read_fn(opened);
+                    os::prepared_publication_test_hooks::run(
+                        os::prepared_publication_test_hooks::Stage::MetadataReadAfterRead,
+                        &path,
+                    );
+                    Ok(result)
+                }
+                #[cfg(not(all(any(test, feature = "test-util"), not(windows))))]
                 Ok(read_fn(root.open_readonly(&path)))
             })
             .await

crates/ecstore/src/disk/os.rs

@@ -333,6 +333,8 @@ pub(crate) mod prepared_publication_test_hooks {
         PreparedRename,
         Rename,
         Remove,
+        MetadataReadAfterOpen,
+        MetadataReadAfterRead,
         #[cfg(test)]
         Rollback,
         #[cfg(test)]
@@ -457,6 +459,8 @@ pub enum LocalPublicationStage {
     PreparedRename,
     Rename,
     Remove,
+    MetadataReadAfterOpen,
+    MetadataReadAfterRead,
 }
 
 #[cfg(all(feature = "test-util", not(windows)))]
@@ -469,6 +473,8 @@ impl LocalPublicationPause {
             LocalPublicationStage::PreparedRename => prepared_publication_test_hooks::Stage::PreparedRename,
             LocalPublicationStage::Rename => prepared_publication_test_hooks::Stage::Rename,
             LocalPublicationStage::Remove => prepared_publication_test_hooks::Stage::Remove,
+            LocalPublicationStage::MetadataReadAfterOpen => prepared_publication_test_hooks::Stage::MetadataReadAfterOpen,
+            LocalPublicationStage::MetadataReadAfterRead => prepared_publication_test_hooks::Stage::MetadataReadAfterRead,
         };
         let (entered_tx, entered) = oneshot::channel();
         let (release, release_rx) = std::sync::mpsc::channel::<()>();

crates/scanner/src/error.rs

@@ -53,6 +53,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

@@ -50,9 +50,9 @@ use storage_api::owner::{
 };
 #[cfg(test)]
 use storage_api::owner::{
-    EcstoreDiskOption, EcstoreDiskStore, EcstoreEndpoint, EcstoreEndpointServerPools, EcstoreEndpoints, EcstoreInstanceContext,
-    EcstorePoolEndpoints, ecstore_config_init, ecstore_init_bucket_metadata_sys, ecstore_init_local_disks_with_instance_ctx,
-    ecstore_new_disk,
+    EcstoreDiskOption, EcstoreDiskStore, EcstoreEndpoint, EcstoreEndpointServerPools, EcstoreEndpoints, EcstoreExpiryState,
+    EcstoreInstanceContext, EcstorePoolEndpoints, ecstore_config_init, ecstore_init_bucket_metadata_sys,
+    ecstore_init_local_disks_with_instance_ctx, ecstore_new_disk,
 };
 use tokio_util::sync::CancellationToken;
 use uuid::Uuid;
@@ -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);
@@ -678,8 +678,37 @@ fn reset_tier_name_cache_for_test() {
     TIER_REGISTRY_GENERATION.store(0, Ordering::Relaxed);
 }
 
-pub(crate) async fn enqueue_runtime_free_version(oi: ScannerObjectInfo) {
-    ecstore_expiry_state_handle().write().await.enqueue_free_version(oi);
+// Only a task running an isolated unit test can override these handles.
+#[cfg(test)]
+struct ExpiryLockTestContext {
+    state: Arc<tokio::sync::RwLock<EcstoreExpiryState>>,
+    store: Option<Arc<ECStore>>,
+    completed_incarnation_reads: Arc<std::sync::atomic::AtomicUsize>,
+}
+
+#[cfg(test)]
+tokio::task_local! {
+    static EXPIRY_LOCK_TEST_CONTEXT: ExpiryLockTestContext;
+}
+
+pub(crate) async fn enqueue_runtime_free_version(oi: ScannerObjectInfo, cancellation: Option<&CancellationToken>) {
+    #[cfg(test)]
+    let state = EXPIRY_LOCK_TEST_CONTEXT
+        .try_with(|context| context.state.clone())
+        .unwrap_or_else(|_| ecstore_expiry_state_handle());
+    #[cfg(not(test))]
+    let state = ecstore_expiry_state_handle();
+    let state = match cancellation {
+        Some(cancel) => tokio::select! {
+            biased;
+            _ = cancel.cancelled() => return,
+            state = state.write() => state,
+        },
+        None => state.write().await,
+    };
+    if !cancellation.is_some_and(CancellationToken::is_cancelled) {
+        state.enqueue_free_version(oi);
+    }
 }
 
 pub(crate) async fn enqueue_runtime_newer_noncurrent(
@@ -687,20 +716,50 @@ pub(crate) async fn enqueue_runtime_newer_noncurrent(
     to_delete_objs: Vec<ObjectToDelete>,
     event: Event,
     src: &LcEventSrc,
+    cancellation: Option<&CancellationToken>,
 ) -> bool {
-    let Some(store) = ecstore_resolve_object_store_handle() else {
+    if cancellation.is_some_and(CancellationToken::is_cancelled) {
+        return false;
+    }
+    #[cfg(test)]
+    let store = EXPIRY_LOCK_TEST_CONTEXT
+        .try_with(|context| context.store.clone())
+        .unwrap_or_else(|_| ecstore_resolve_object_store_handle());
+    #[cfg(not(test))]
+    let store = ecstore_resolve_object_store_handle();
+    let Some(store) = store else {
         return false;
     };
     let Ok(bucket_incarnation_id) = store.bucket_incarnation_id(bucket).await else {
         return false;
     };
-    ecstore_expiry_state_handle().write().await.enqueue_by_newer_noncurrent(
-        bucket,
-        to_delete_objs,
-        event,
-        src,
-        bucket_incarnation_id,
-    )
+    #[cfg(test)]
+    let _ = EXPIRY_LOCK_TEST_CONTEXT.try_with(|context| {
+        context
+            .completed_incarnation_reads
+            .fetch_add(1, std::sync::atomic::Ordering::SeqCst);
+    });
+    if cancellation.is_some_and(CancellationToken::is_cancelled) {
+        return false;
+    }
+    #[cfg(test)]
+    let state = EXPIRY_LOCK_TEST_CONTEXT
+        .try_with(|context| context.state.clone())
+        .unwrap_or_else(|_| ecstore_expiry_state_handle());
+    #[cfg(not(test))]
+    let state = ecstore_expiry_state_handle();
+    let mut state = match cancellation {
+        Some(cancel) => tokio::select! {
+            biased;
+            _ = cancel.cancelled() => return false,
+            state = state.write() => state,
+        },
+        None => state.write().await,
+    };
+    if cancellation.is_some_and(CancellationToken::is_cancelled) {
+        return false;
+    }
+    state.enqueue_by_newer_noncurrent(bucket, to_delete_objs, event, src, bucket_incarnation_id)
 }
 
 pub(crate) async fn queue_replication_heal(

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

@@ -1006,6 +1006,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

@@ -43,6 +43,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::{
@@ -795,6 +796,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,
@@ -828,6 +830,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 {
@@ -998,12 +1002,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.
@@ -1923,7 +1928,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()
@@ -1961,7 +1971,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;
@@ -2072,7 +2082,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!(
@@ -2123,7 +2138,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!(
@@ -2274,22 +2294,30 @@ 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 size_result = 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?;
+                if self.node_admission.is_some() && ctx.is_cancelled() {
+                    return Err(ScannerError::Other("Operation cancelled".to_string()));
+                }
+                let sz = match size_result {
                     Ok(sz) => sz,
                     Err(e) => {
                         let failure_action = classify_get_size_failure(&item, &e);
@@ -2606,7 +2634,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);
@@ -2847,19 +2875,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;
@@ -2985,6 +3017,7 @@ impl FolderScanner {
                                 finished_closed = true;
                                 continue;
                             };
+                            finished_closed = true;
                             if disk_errors_are_only_missing_paths(&errs) {
                                 debug!(
                                     target: "rustfs::scanner::folder",
@@ -3006,7 +3039,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;
@@ -3024,7 +3059,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 {
@@ -3043,7 +3081,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);
@@ -3217,6 +3255,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,
         usage_updates,
@@ -3244,6 +3283,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();
@@ -3333,7 +3374,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;
@@ -3408,6 +3449,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)
@@ -3427,3 +3481,15 @@ pub(crate) use tests::checkpoint_fixture::run_checkpoint_fixture;
 
 #[cfg(test)]
 mod tests;
+
+#[cfg(test)]
+tokio::task_local! {
+    pub(crate) static TEST_SIZE_DISPATCH_BOUNDARIES: tokio::sync::mpsc::UnboundedSender<&'static str>;
+}
+
+#[cfg(test)]
+pub(crate) fn observe_test_size_dispatch_boundary(stage: &'static str) -> bool {
+    TEST_SIZE_DISPATCH_BOUNDARIES
+        .try_with(|sender| sender.send(stage).is_ok())
+        .unwrap_or(false)
+}

crates/scanner/src/scanner_folder/item_actions.rs

@@ -628,6 +628,22 @@ impl ScannerItem {
         tier_names: &[String],
         size_summary: &mut SizeSummary,
     ) {
+        self.apply_actions_cancellable(object_infos, lock_retention, versioning_config, tier_names, size_summary, None)
+            .await;
+    }
+
+    pub(crate) async fn apply_actions_cancellable(
+        &mut self,
+        object_infos: Vec<ObjectInfo>,
+        lock_retention: Option<Arc<ObjectLockConfiguration>>,
+        versioning_config: VersioningConfiguration,
+        tier_names: &[String],
+        size_summary: &mut SizeSummary,
+        cancellation: Option<&CancellationToken>,
+    ) {
+        if cancellation.is_some_and(CancellationToken::is_cancelled) {
+            return;
+        }
         let object_path = self.object_path();
         if object_infos.is_empty() {
             debug!(
@@ -671,6 +687,9 @@ impl ScannerItem {
         let Some(lifecycle) = self.lifecycle.clone() else {
             let mut cumulative_size: i64 = 0;
             for (oi, resolved_size) in object_infos.iter().zip(resolved_sizes.iter()) {
+                if cancellation.is_some_and(CancellationToken::is_cancelled) {
+                    return;
+                }
                 let accounting_size = match resolved_size {
                     SizeResolution::Known { logical, .. } => *logical,
                     // A valid compressed legacy sentinel has no logical size,
@@ -678,7 +697,7 @@ impl ScannerItem {
                     // physical size is only an input to those operations; it
                     // is not folded into the logical total below.
                     SizeResolution::Unknown { physical, .. } => {
-                        self.heal_actions(oi, *physical, size_summary).await;
+                        self.heal_actions(oi, *physical, size_summary, cancellation).await;
                         size_summary.actions_accounting_unknown(oi);
                         continue;
                     }
@@ -688,13 +707,16 @@ impl ScannerItem {
                     }
                 };
 
-                let size = self.heal_actions(oi, accounting_size, size_summary).await;
+                let size = self.heal_actions(oi, accounting_size, size_summary, cancellation).await;
 
                 size_summary.actions_accounting(oi, size, accounting_size);
 
                 cumulative_size = cumulative_size.saturating_add(size);
             }
 
+            if cancellation.is_some_and(CancellationToken::is_cancelled) {
+                return;
+            }
             self.alert_excessive_versions(object_infos.len(), cumulative_size);
 
             debug!(
@@ -736,6 +758,10 @@ impl ScannerItem {
             }
         };
 
+        if cancellation.is_some_and(CancellationToken::is_cancelled) {
+            return;
+        }
+
         // Every version handed to the evaluator counts as an ILM-checked
         // version, whether or not a rule matched (NoneAction included). This is
         // reached only for buckets with lifecycle rules; it feeds
@@ -754,6 +780,9 @@ impl ScannerItem {
             .all(|candidate| Self::tier_is_known(candidate, tier_names));
         'eventLoop: {
             for (i, event) in events.iter().enumerate() {
+                if cancellation.is_some_and(CancellationToken::is_cancelled) {
+                    return;
+                }
                 let oi = &object_infos[i];
                 let known_size = resolved_sizes[i].known_size();
                 if has_corrupt_size
@@ -772,7 +801,7 @@ impl ScannerItem {
                     // accounting. Size-filtered or deferred events remain
                     // pending, so retain the version-only physical counters.
                     if let SizeResolution::Unknown { physical, .. } = &resolved_sizes[i] {
-                        self.heal_actions(oi, *physical, size_summary).await;
+                        self.heal_actions(oi, *physical, size_summary, cancellation).await;
                         size_summary.actions_accounting_unknown(oi);
                     }
                     continue;
@@ -829,7 +858,7 @@ impl ScannerItem {
                                     noncurrent_unknown.push(oi);
                                 } else {
                                     if let SizeResolution::Unknown { physical, .. } = &resolved_sizes[i] {
-                                        self.heal_actions(oi, *physical, size_summary).await;
+                                        self.heal_actions(oi, *physical, size_summary, cancellation).await;
                                     }
                                     size_summary.actions_accounting_unknown(oi);
                                 }
@@ -846,7 +875,7 @@ impl ScannerItem {
                             }
                             IlmAction::NoneAction | IlmAction::ActionCount => {
                                 if let SizeResolution::Unknown { physical, .. } = &resolved_sizes[i] {
-                                    self.heal_actions(oi, *physical, size_summary).await;
+                                    self.heal_actions(oi, *physical, size_summary, cancellation).await;
                                 }
                                 size_summary.actions_accounting_unknown(oi);
                             }
@@ -864,7 +893,7 @@ impl ScannerItem {
                 // side-effecting tier actions fail closed until the registry
                 // recognizes the source again.
                 if Self::action_blocked_by_unknown_tier(event.action, oi, all_versions_known, tier_names, &event.storage_class) {
-                    size = self.heal_actions(oi, actual_size, size_summary).await;
+                    size = self.heal_actions(oi, actual_size, size_summary, cancellation).await;
                     size_summary.actions_accounting(oi, size, actual_size);
                     cumulative_size += size;
                     continue;
@@ -973,7 +1002,7 @@ impl ScannerItem {
                     }
 
                     IlmAction::NoneAction | IlmAction::ActionCount => {
-                        size = self.heal_actions(oi, actual_size, size_summary).await;
+                        size = self.heal_actions(oi, actual_size, size_summary, cancellation).await;
                     }
                 }
 
@@ -984,6 +1013,9 @@ impl ScannerItem {
             }
         }
 
+        if cancellation.is_some_and(CancellationToken::is_cancelled) {
+            return;
+        }
         if !to_delete_objs.is_empty()
             && let Some(event) = noncurrent_events.first().cloned()
         {
@@ -991,7 +1023,8 @@ impl ScannerItem {
             let count = u64::try_from(to_delete_objs.len()).unwrap_or(u64::MAX);
             let done_ilm = Metrics::time_ilm(action);
             let trace_started_at = trace_start_instant();
-            let queued = enqueue_runtime_newer_noncurrent(&self.bucket, to_delete_objs, event, &LcEventSrc::Scanner).await;
+            let queued =
+                enqueue_runtime_newer_noncurrent(&self.bucket, to_delete_objs, event, &LcEventSrc::Scanner, cancellation).await;
             if let Some(trace_started_at) = trace_started_at {
                 let state = if queued { "queued" } else { "not_queued" };
                 trace_emit(|| {
@@ -1022,15 +1055,28 @@ impl ScannerItem {
                 }
             }
         }
-        self.alert_excessive_versions(remaining_versions, cumulative_size);
+        if !cancellation.is_some_and(CancellationToken::is_cancelled) {
+            self.alert_excessive_versions(remaining_versions, cumulative_size);
+        }
     }
 
-    pub(super) async fn heal_actions(&mut self, oi: &ObjectInfo, actual_size: i64, size_summary: &mut SizeSummary) -> i64 {
+    pub(super) async fn heal_actions(
+        &mut self,
+        oi: &ObjectInfo,
+        actual_size: i64,
+        size_summary: &mut SizeSummary,
+        cancellation: Option<&CancellationToken>,
+    ) -> i64 {
+        if cancellation.is_some_and(CancellationToken::is_cancelled) {
+            return actual_size;
+        }
         if self.heal_enabled {
             self.enqueue_heal(oi).await;
         }
 
-        self.heal_replication(oi, size_summary).await;
+        if !cancellation.is_some_and(CancellationToken::is_cancelled) {
+            self.heal_replication(oi, size_summary).await;
+        }
 
         actual_size
     }

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;
 mod usage_updates;
 
 /// Reset the process-global alert cooldown map; test-only.
@@ -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,
         usage_updates: None,
@@ -394,6 +397,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)
@@ -2262,7 +2266,7 @@ async fn test_heal_actions_returns_actual_size_without_inline_heal() {
     };
     let mut size_summary = SizeSummary::default();
 
-    let size = item.heal_actions(&object_info, 123, &mut size_summary).await;
+    let size = item.heal_actions(&object_info, 123, &mut size_summary, None).await;
     assert_eq!(size, 123);
 }
 

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] {
+        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_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

@@ -1090,7 +1090,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() {
@@ -1320,9 +1320,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(),
@@ -1333,7 +1334,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);
@@ -1341,6 +1342,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/io_disk.rs

@@ -38,6 +38,13 @@ pub(super) fn tier_stats_template(tier_names: &[String]) -> HashMap<String, Tier
     tier_stats
 }
 
+fn ensure_scanner_node_operation_active(cancellation: Option<&CancellationToken>) -> Result<()> {
+    if cancellation.is_some_and(CancellationToken::is_cancelled) {
+        return Err(StorageError::other("Scanner operation cancelled"));
+    }
+    Ok(())
+}
+
 #[async_trait::async_trait]
 impl ScannerIODisk for Disk {
     async fn get_size(&self, item: ScannerItem) -> Result<SizeSummary> {
@@ -45,14 +52,18 @@ impl ScannerIODisk for Disk {
     }
 
     async fn get_size_with_tier_names(&self, mut item: ScannerItem, tier_names: &[String]) -> Result<SizeSummary> {
+        let cancellation = crate::workload_admission::scanner_node_operation_cancellation();
         let done_object = Metrics::time(Metric::ScanObject);
 
         if !is_xl_meta_path(&item.path) {
             return Err(StorageError::other(SCANNER_SKIP_FILE_ERROR.to_string()));
         }
 
+        ensure_scanner_node_operation_active(cancellation.as_ref())?;
         let metadata_object_path = item.object_path();
-        let data = match self.read_metadata(&item.bucket, &metadata_object_path).await {
+        let metadata_result = self.read_metadata(&item.bucket, &metadata_object_path).await;
+        ensure_scanner_node_operation_active(cancellation.as_ref())?;
+        let data = match metadata_result {
             Ok(data) => data,
             Err(e) if DiskError::is_err_object_not_found(&e) || DiskError::is_err_version_not_found(&e) => {
                 return Err(StorageError::other(SCANNER_SKIP_FILE_ERROR.to_string()));
@@ -86,7 +97,10 @@ impl ScannerIODisk for Disk {
         // (which used to query it a second time). On failure keep the
         // historical fallback: default configuration (versioned = false) plus
         // the warn that `apply_actions` used to emit.
-        let versioning_config = match BucketVersioningSys::get(&item.bucket).await {
+        ensure_scanner_node_operation_active(cancellation.as_ref())?;
+        let versioning_result = BucketVersioningSys::get(&item.bucket).await;
+        ensure_scanner_node_operation_active(cancellation.as_ref())?;
+        let versioning_config = match versioning_result {
             Ok(versioning_config) => versioning_config,
             Err(_) => {
                 warn!(
@@ -122,7 +136,9 @@ impl ScannerIODisk for Disk {
             ..Default::default()
         };
 
+        ensure_scanner_node_operation_active(cancellation.as_ref())?;
         let lock_config = object_lock_config_for_scanner_item(&item).await;
+        ensure_scanner_node_operation_active(cancellation.as_ref())?;
 
         // Count every version this object contributes to the scan, independent
         // of any lifecycle configuration, so scan-coverage metrics stay honest
@@ -130,17 +146,33 @@ impl ScannerIODisk for Disk {
         // `object_infos`.
         global_metrics().record_scanner_versions_scanned(object_infos.len() as u64);
 
-        item.apply_actions(object_infos, lock_config, versioning_config, tier_names, &mut size_summary)
-            .await;
+        #[cfg(test)]
+        let _ = crate::scanner_folder::observe_test_size_dispatch_boundary("apply_actions");
+        item.apply_actions_cancellable(
+            object_infos,
+            lock_config,
+            versioning_config,
+            tier_names,
+            &mut size_summary,
+            cancellation.as_ref(),
+        )
+        .await;
+        ensure_scanner_node_operation_active(cancellation.as_ref())?;
 
         if !free_version_infos.is_empty() {
             for oi in free_version_infos {
+                ensure_scanner_node_operation_active(cancellation.as_ref())?;
                 if ScannerItem::tier_is_known(&oi, tier_names) {
-                    enqueue_runtime_free_version(oi).await;
+                    #[cfg(test)]
+                    if crate::scanner_folder::observe_test_size_dispatch_boundary("free_version_dispatch") {
+                        continue;
+                    }
+                    enqueue_runtime_free_version(oi, cancellation.as_ref()).await;
                 }
             }
         }
 
+        ensure_scanner_node_operation_active(cancellation.as_ref())?;
         done_object();
 
         Ok(size_summary)

crates/scanner/src/scanner_io/tests.rs

@@ -46,7 +46,9 @@ use uuid::Uuid;
 
 mod checkpoint_coverage_identity;
 mod collector_lifecycle;
+mod expiry_lock;
 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/expiry_lock.rs

@@ -0,0 +1,181 @@
+// 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.
+
+use super::*;
+use crate::storage_api::owner::EcstoreExpiryState;
+use std::future::Future;
+use std::pin::Pin;
+use std::sync::atomic::{AtomicUsize, Ordering};
+use std::task::Poll;
+use std::time::Duration;
+
+// Return Some(result) if the operation ended too early, None only if it was
+// actually polled pending after the prerequisite read had finished.
+async fn observe_queued<F: Future>(
+    mut operation: Pin<&mut F>,
+    read_complete: impl Fn() -> bool,
+    observation: Duration,
+) -> std::result::Result<Option<F::Output>, tokio::time::error::Elapsed> {
+    tokio::time::timeout(
+        observation,
+        std::future::poll_fn(|cx| match operation.as_mut().poll(cx) {
+            Poll::Ready(value) => Poll::Ready(Some(value)),
+            Poll::Pending if read_complete() => Poll::Ready(None),
+            Poll::Pending => Poll::Pending,
+        }),
+    )
+    .await
+}
+
+#[tokio::test]
+#[serial]
+async fn free_version_cancelled_while_waiting_for_expiry_write_lock_never_queues() {
+    let state = EcstoreExpiryState::new_with_unconsumed_worker_channel(4);
+    let observation = Duration::from_secs(rustfs_config::DEFAULT_DRIVE_METADATA_TIMEOUT_SECS);
+    let context = crate::ExpiryLockTestContext {
+        state: state.clone(),
+        store: None,
+        completed_incarnation_reads: Arc::new(AtomicUsize::new(0)),
+    };
+    crate::EXPIRY_LOCK_TEST_CONTEXT
+        .scope(context, async {
+            let mut object = crate::ScannerObjectInfo::default();
+            object.bucket = "expiry-lock-bucket".to_string();
+            object.name = "free-object".to_string();
+            let cancellation = CancellationToken::new();
+            let guard = state.write().await;
+            let mut pending = Box::pin(crate::enqueue_runtime_free_version(object.clone(), Some(&cancellation)));
+            let first = observe_queued(pending.as_mut(), || true, observation).await;
+            let observed_wait = matches!(&first, Ok(None));
+            let already_finished = matches!(&first, Ok(Some(())));
+            cancellation.cancel();
+            let cancelled_while_locked = if already_finished {
+                false
+            } else {
+                tokio::time::timeout(observation, pending.as_mut()).await.is_ok()
+            };
+            let pending_while_locked = guard.pending_tasks();
+            drop(guard);
+            if !already_finished && !cancelled_while_locked {
+                pending.as_mut().await;
+            }
+            drop(pending);
+            let pending_after_cancel = state.read().await.pending_tasks();
+
+            // Positive control queues a real free-version task while using the same
+            // implementation and isolated, unconsumed worker channel.
+            let mut positive = Box::pin(crate::enqueue_runtime_free_version(object, None));
+            let positive_in_time = tokio::time::timeout(observation, positive.as_mut()).await.is_ok();
+            if !positive_in_time {
+                positive.as_mut().await;
+            }
+            drop(positive);
+            let pending_after_control = state.read().await.pending_tasks();
+            assert!(observed_wait, "must poll the real enqueue while the write lock is held");
+            assert!(cancelled_while_locked, "cancellation must finish before the lock is released");
+            assert_eq!(pending_while_locked, 0);
+            assert_eq!(pending_after_cancel, 0);
+            assert!(positive_in_time, "positive control must finish within the observation limit");
+            assert_eq!(pending_after_control, 1, "positive control must queue a task");
+        })
+        .await;
+}
+
+#[tokio::test]
+#[serial]
+async fn newer_noncurrent_cancelled_after_real_read_while_waiting_for_expiry_write_lock_never_queues() {
+    let (_temp_dir, store) = setup_local_scanner_store(1, 4).await;
+    let bucket = "expiry-lock-bucket";
+    store
+        .make_bucket(bucket, &MakeBucketOptions::default())
+        .await
+        .expect("create bucket");
+    let expected_incarnation = store.bucket_incarnation_id(bucket).await.expect("real incarnation metadata");
+    assert!(!expected_incarnation.is_nil());
+    let state = EcstoreExpiryState::new_with_unconsumed_worker_channel(4);
+    let reads = Arc::new(AtomicUsize::new(0));
+    let context = crate::ExpiryLockTestContext {
+        state: state.clone(),
+        store: Some(store.clone()),
+        completed_incarnation_reads: reads.clone(),
+    };
+    let observation = Duration::from_secs(rustfs_config::DEFAULT_DRIVE_METADATA_TIMEOUT_SECS);
+    crate::EXPIRY_LOCK_TEST_CONTEXT
+        .scope(context, async {
+            let versions = vec![crate::ScannerObjectToDelete {
+                object_name: "noncurrent-object".to_string(),
+                version_id: Some(Uuid::new_v4()),
+                ..Default::default()
+            }];
+            let event = crate::Event::default();
+            let src = crate::LcEventSrc::Scanner;
+            let cancellation = CancellationToken::new();
+            let guard = state.write().await;
+            let mut pending = Box::pin(crate::enqueue_runtime_newer_noncurrent(
+                bucket,
+                versions.clone(),
+                event.clone(),
+                &src,
+                Some(&cancellation),
+            ));
+            let first = observe_queued(pending.as_mut(), || reads.load(Ordering::SeqCst) == 1, observation).await;
+            let observed_wait_after_read = matches!(&first, Ok(None));
+            let mut result = match first {
+                Ok(Some(value)) => Some(value),
+                _ => None,
+            };
+            cancellation.cancel();
+            let cancelled_while_locked = if result.is_some() {
+                false
+            } else {
+                match tokio::time::timeout(observation, pending.as_mut()).await {
+                    Ok(value) => {
+                        result = Some(value);
+                        true
+                    }
+                    Err(_) => false,
+                }
+            };
+            let pending_while_locked = guard.pending_tasks();
+            drop(guard);
+            let result = match result {
+                Some(value) => value,
+                None => pending.as_mut().await,
+            };
+            drop(pending);
+            let pending_after_cancel = state.read().await.pending_tasks();
+            let mut positive = Box::pin(crate::enqueue_runtime_newer_noncurrent(bucket, versions, event, &src, None));
+            let bounded_positive = tokio::time::timeout(observation, positive.as_mut()).await;
+            let positive_in_time = bounded_positive.is_ok();
+            let positive_result = match bounded_positive {
+                Ok(value) => value,
+                Err(_) => positive.as_mut().await,
+            };
+            drop(positive);
+            let pending_after_control = state.read().await.pending_tasks();
+            assert!(
+                observed_wait_after_read,
+                "real incarnation read must finish before waiting on the held lock"
+            );
+            assert!(cancelled_while_locked, "cancellation must finish before the lock is released");
+            assert!(!result);
+            assert_eq!(pending_while_locked, 0);
+            assert_eq!(pending_after_cancel, 0);
+            assert!(positive_in_time, "positive control must finish within the observation limit");
+            assert!(positive_result, "positive control must enqueue the nonempty version batch");
+            assert_eq!(pending_after_control, 1);
+            assert_eq!(reads.load(Ordering::SeqCst), 2);
+        })
+        .await;
+}

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/storage_api.rs

@@ -39,6 +39,8 @@ pub(crate) use rustfs_ecstore::api::data_usage::{NativeScannerPauseBacklogWriteF
 
 pub(crate) use rustfs_ecstore::api::bucket::bucket_target_sys::BucketTargetSys as EcstoreBucketTargetSys;
 pub(crate) use rustfs_ecstore::api::bucket::lifecycle::bucket_lifecycle_audit::LcEventSrc as EcstoreLcEventSrc;
+#[cfg(test)]
+pub(crate) use rustfs_ecstore::api::bucket::lifecycle::bucket_lifecycle_ops::ExpiryState as EcstoreExpiryState;
 pub(crate) use rustfs_ecstore::api::bucket::lifecycle::bucket_lifecycle_ops::{
     apply_expiry_rule as ecstore_apply_expiry_rule, apply_transition_rule as ecstore_apply_transition_rule,
     lifecycle_version_delete_target as ecstore_lifecycle_version_delete_target,
@@ -161,6 +163,8 @@ pub(crate) mod ecstore_diagnostic {
 pub(crate) type EcstoreHealResultItem = <EcstoreStore as storage_contracts::HealOperations>::HealResultItem;
 
 pub(crate) mod owner {
+    #[cfg(test)]
+    pub(crate) use super::EcstoreExpiryState;
     #[cfg(test)]
     pub(crate) use super::ecstore_set_object_store_resolver;
     #[cfg(test)]

crates/scanner/src/workload_admission.rs

@@ -12,9 +12,143 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::sync::{Arc, LazyLock, RwLock};
+use std::sync::atomic::{AtomicBool, Ordering};
+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;
+    static SCANNER_NODE_OPERATION_CANCEL: CancellationToken;
+}
+
+/// Cancellation for the currently admitted scanner operation, when present.
+pub(crate) fn scanner_node_operation_cancellation() -> Option<CancellationToken> {
+    SCANNER_NODE_OPERATION_CANCEL.try_with(Clone::clone).ok()
+}
+
+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>;
+    pub(crate) static TEST_SCANNER_AFTER_NODE_ACQUIRE: Arc<dyn Fn() + Send + Sync>;
+}
+
+/// 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);
+    };
+    let operation = SCANNER_NODE_OPERATION_CANCEL.scope(cancel.clone(), operation);
+    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?,
+            }
+        }
+    };
+    #[cfg(test)]
+    let _ = TEST_SCANNER_AFTER_NODE_ACQUIRE.try_with(|hook| hook());
+    let operation_started = AtomicBool::new(false);
+    tokio::pin!(operation);
+    let observed = std::future::poll_fn(|context| {
+        operation_started.store(true, Ordering::Relaxed);
+        std::future::Future::poll(operation.as_mut(), context)
+    });
+    tokio::pin!(observed);
+    let result = tokio::select! {
+        biased;
+        _ = cancel.cancelled() => {
+            // The node slot belongs to the waiter, not the native operation.
+            drop(permit);
+            if !operation_started.load(Ordering::Relaxed) {
+                return Err(ScannerError::Other("Operation cancelled".to_string()));
+            }
+            // Keep polling the started operation under its existing locks.
+            return Ok(observed.await);
+        }
+        result = &mut observed => result,
+    };
+    if result.is_ok() && !cancel.is_cancelled() {
+        permit.complete();
+    }
+    Ok(result)
+}
 
 type WorkloadSnapshotProviderRef = Arc<dyn WorkloadAdmissionSnapshotProvider + Send + Sync>;
 

docs/operations/scanner-runtime-controls.md

@@ -529,17 +529,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"):