perf(ecstore): bound and isolate cleanup execution (#8151)

* perf(ecstore): bound and isolate cleanup execution

* fix(ecstore): preserve cleanup cancellation and lease release

crates/ecstore/Cargo.toml

@@ -200,8 +200,8 @@ md-5.workspace = true
 memmap2 = { workspace = true }
 libc.workspace = true
 # "process" adds getrlimit for the io_uring fd-cache RLIMIT_NOFILE gate
-# (backlog#1178); "fs" comes from the workspace default.
-rustix = { workspace = true, features = ["process", "fs"] }
+# (backlog#1178); "thread" provides cleanup affinity; "fs" is a workspace default.
+rustix = { workspace = true, features = ["process", "fs", "thread"] }
 rustfs-madmin.workspace = true
 reqwest = { workspace = true }
 aws-sdk-s3 = { workspace = true, default-features = false, features = ["sigv4a", "default-https-client", "rt-tokio"] }

crates/ecstore/src/disk/disk_store.rs

@@ -12,6 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use crate::disk::cleanup_runtime;
 use crate::disk::{
     CheckPartsResp, DataDirDeleteStatus, DeleteOptions, DiskAPI, DiskError, DiskInfo, DiskInfoOptions, DiskLocation, Endpoint,
     Error, FileInfoVersions, MmapCopyStageMetrics, ReadMultipleReq, ReadMultipleResp, ReadOptions, RenameDataResp, Result,
@@ -364,10 +365,10 @@ impl LocalDiskWrapper {
         opts: DeleteOptions,
         namespace_owner: Option<Arc<dyn Send + Sync>>,
     ) -> Result<()> {
-        self.track_disk_health_mutation(
+        self.track_cleanup_mutation(
             "delete",
             DiskMetricMutation::Delete,
-            || async { Box::pin(self.disk.delete_with_namespace_owner(volume, path, opts, namespace_owner)).await },
+            || async { Box::pin(self.cleanup_delete(volume, path, opts, namespace_owner)).await },
             get_max_timeout_duration(),
         )
         .await
@@ -1339,6 +1340,28 @@ impl LocalDiskWrapper {
         )
     }
 
+    async fn cleanup_delete(
+        &self,
+        volume: &str,
+        path: &str,
+        options: DeleteOptions,
+        owner: Option<Arc<dyn Send + Sync>>,
+    ) -> Result<()> {
+        if !cleanup_runtime::enabled()? {
+            return self.disk.delete_with_namespace_owner(volume, path, options, owner).await;
+        }
+        let disk = self.disk.clone();
+        let volume = volume.to_owned();
+        let path = path.to_owned();
+        let guarded = owner.is_some();
+        let operation = async move { disk.delete_with_namespace_owner(&volume, &path, options, owner).await };
+        if guarded {
+            cleanup_runtime::guarded_disk_operation(operation).await
+        } else {
+            cleanup_runtime::disk_operation(operation).await
+        }
+    }
+
     /// Run a delete under an owned coordinator task when a publication guard
     /// is present. This keeps the guard alive if the RPC waiter is cancelled
     /// while the local namespace mutation is still in progress.
@@ -1357,12 +1380,13 @@ impl LocalDiskWrapper {
         } else {
             get_max_timeout_duration()
         };
+        let cleanup_owner = external_guard.clone();
         run_owned_mutation(external_guard, move || async move {
             operation
-                .track_disk_health_mutation(
+                .track_cleanup_mutation(
                     "delete",
                     DiskMetricMutation::Delete,
-                    || async { operation.disk.delete(&volume, &path, options).await },
+                    || async { operation.cleanup_delete(&volume, &path, options, cleanup_owner).await },
                     timeout_duration,
                 )
                 .await
@@ -1821,6 +1845,31 @@ impl LocalDiskWrapper {
         self.track_disk_health_with_op("unknown", operation, timeout_duration).await
     }
 
+    /// Cleanup latency includes waiting behind a node-wide execution budget,
+    /// so a waiter deadline alone is not evidence that this disk is faulty.
+    /// The active disk health probe retains its original timeout policy.
+    async fn track_cleanup_mutation<T, F, Fut>(
+        &self,
+        op: &'static str,
+        mutation: DiskMetricMutation,
+        operation: F,
+        timeout_duration: Duration,
+    ) -> Result<T>
+    where
+        F: FnOnce() -> Fut,
+        Fut: std::future::Future<Output = Result<T>>,
+    {
+        // Runtime initialization must not prevent control-plane release of
+        // reader tokens. Disk I/O entries still propagate configuration errors.
+        let action = if cleanup_runtime::enabled().unwrap_or(false) {
+            TimeoutHealthAction::IgnoreFailure
+        } else {
+            TimeoutHealthAction::MarkFailure
+        };
+        self.track_disk_health_with_op_timeout_action_and_mutation(op, operation, timeout_duration, action, mutation)
+            .await
+    }
+
     async fn track_disk_health_mutation<T, F, Fut>(
         &self,
         op: &'static str,
@@ -2236,9 +2285,37 @@ impl DiskAPI for LocalDiskWrapper {
     }
 
     async fn release_snapshot_lease(&self, volume: &str, path: &str, token: SnapshotLeaseToken) -> Result<()> {
-        self.track_disk_health_with_op(
+        // Quota fence release is control-plane work and may wait for in-flight
+        // mutations. Keep its original waiter lifecycle instead of creating
+        // detached cleanup tasks for repeated fence-release RPCs.
+        if volume == super::RUSTFS_META_BUCKET && super::is_quota_mutation_fence_path(path) {
+            return self
+                .track_disk_health_with_op(
+                    "release_snapshot_lease",
+                    || async { self.disk.release_snapshot_lease(volume, path, token).await },
+                    get_max_timeout_duration(),
+                )
+                .await;
+        }
+        self.track_cleanup_mutation(
             "release_snapshot_lease",
-            || async { self.disk.release_snapshot_lease(volume, path, token).await },
+            DiskMetricMutation::None,
+            || async {
+                if self.disk.release_snapshot_lease_without_cleanup(volume, path, token).await {
+                    return Ok(());
+                }
+                if !cleanup_runtime::enabled().unwrap_or(false) {
+                    // The raw release removes the token before trying the
+                    // reclaim budget. Initialization failure therefore keeps
+                    // the cleanup intent and data, without leaking a reader
+                    // or falling back to unbudgeted filesystem deletion.
+                    return self.disk.release_snapshot_lease(volume, path, token).await;
+                }
+                let disk = self.disk.clone();
+                let volume = volume.to_owned();
+                let path = path.to_owned();
+                cleanup_runtime::release_lease(async move { disk.release_snapshot_lease(&volume, &path, token).await }).await
+            },
             get_max_timeout_duration(),
         )
         .await
@@ -2263,10 +2340,22 @@ impl DiskAPI for LocalDiskWrapper {
         } else {
             get_max_timeout_duration()
         };
-        self.track_disk_health_mutation(
+        self.track_cleanup_mutation(
             "delete_data_dir",
             DiskMetricMutation::Delete,
-            || async { self.disk.delete_data_dir(volume, path, opts).await },
+            || async {
+                if !cleanup_runtime::enabled()? {
+                    return self.disk.delete_data_dir(volume, path, opts).await;
+                }
+                let disk = self.disk.clone();
+                let volume = volume.to_owned();
+                let path = path.to_owned();
+                cleanup_runtime::disk_operation(async move {
+                    let _scope = scope;
+                    disk.delete_data_dir(&volume, &path, opts).await
+                })
+                .await
+            },
             timeout,
         )
         .await
@@ -2491,10 +2580,10 @@ impl DiskAPI for LocalDiskWrapper {
     }
 
     async fn delete(&self, volume: &str, path: &str, opt: DeleteOptions) -> Result<()> {
-        self.track_disk_health_mutation(
+        self.track_cleanup_mutation(
             "delete",
             DiskMetricMutation::Delete,
-            || async { self.disk.delete(volume, path, opt).await },
+            || async { self.cleanup_delete(volume, path, opt, None).await },
             get_max_timeout_duration(),
         )
         .await
@@ -2521,11 +2610,25 @@ impl DiskAPI for LocalDiskWrapper {
     }
 
     async fn write_all(&self, volume: &str, path: &str, data: Bytes) -> Result<()> {
-        self.track_disk_health_mutation(
+        let cleanup = path.rsplit('/').next() == Some(cleanup_runtime::OLD_DATA_CLEANUP_RECEIPT_FILE);
+        self.track_disk_health_with_op_timeout_action_and_mutation(
             "write_all",
-            DiskMetricMutation::Write,
-            || async { self.disk.write_all(volume, path, data).await },
+            || async {
+                if !cleanup || !cleanup_runtime::enabled()? {
+                    return self.disk.write_all(volume, path, data).await;
+                }
+                let disk = self.disk.clone();
+                let volume = volume.to_owned();
+                let path = path.to_owned();
+                cleanup_runtime::disk_operation(async move { disk.write_all(&volume, &path, data).await }).await
+            },
             get_max_timeout_duration(),
+            if cleanup && cleanup_runtime::enabled()? {
+                TimeoutHealthAction::IgnoreFailure
+            } else {
+                TimeoutHealthAction::MarkFailure
+            },
+            DiskMetricMutation::Write,
         )
         .await
     }
@@ -2577,6 +2680,256 @@ mod tests {
         }
     }
 
+    #[tokio::test]
+    async fn invalid_cleanup_config_does_not_leak_snapshot_readers() {
+        const CHILD: &str = "RUSTFS_CLEANUP_INVALID_LEASE_TEST_CHILD";
+        if std::env::var_os(CHILD).is_none() {
+            let output = tokio::task::spawn_blocking(|| {
+                std::process::Command::new(std::env::current_exe().expect("test executable"))
+                    .args([
+                        "--exact",
+                        "disk::disk_store::tests::invalid_cleanup_config_does_not_leak_snapshot_readers",
+                        "--nocapture",
+                    ])
+                    .env(CHILD, "1")
+                    .env("RUSTFS_CLEANUP_ISOLATE_ENABLE", "true")
+                    .env("RUSTFS_CLEANUP_ASYNC_THREADS", "0")
+                    .output()
+                    .expect("run with invalid cleanup configuration")
+            })
+            .await
+            .expect("child process task");
+            assert!(
+                output.status.success(),
+                "{}\n{}",
+                String::from_utf8_lossy(&output.stdout),
+                String::from_utf8_lossy(&output.stderr)
+            );
+            assert!(String::from_utf8_lossy(&output.stdout).contains("1 passed"));
+            return;
+        }
+        let dir = tempfile::tempdir().expect("temporary disk");
+        let endpoint = Endpoint::try_from(dir.path().to_str().expect("UTF-8 path")).expect("endpoint");
+        let disk = Arc::new(LocalDisk::new(&endpoint, false).await.expect("local disk"));
+        let wrapper = LocalDiskWrapper::new(disk.clone(), false);
+        let volume = "invalid-cleanup-config";
+        let path = "object/data";
+        wrapper.make_volume(volume).await.expect("bucket");
+        wrapper
+            .write_all(volume, "object/data/part.1", Bytes::from_static(b"live shard"))
+            .await
+            .expect("write shard");
+        let token = wrapper.acquire_snapshot_lease(volume, path).await.expect("GET lease");
+        wrapper
+            .release_snapshot_lease(volume, path, token)
+            .await
+            .expect("control-only release must not depend on cleanup initialization");
+        assert!(matches!(
+            wrapper.renew_snapshot_lease(volume, path, token).await,
+            Err(DiskError::FileNotFound)
+        ));
+
+        let token = wrapper
+            .acquire_snapshot_lease(volume, path)
+            .await
+            .expect("reader of old data");
+        // Register reclaim intent directly; this only updates the lease
+        // registry while a reader is present and performs no filesystem I/O.
+        assert_eq!(
+            disk.delete_data_dir(
+                volume,
+                path,
+                DeleteOptions {
+                    recursive: true,
+                    ..Default::default()
+                }
+            )
+            .await
+            .expect("defer reclaim"),
+            DataDirDeleteStatus::Deferred
+        );
+        assert!(
+            wrapper.release_snapshot_lease(volume, path, token).await.is_err(),
+            "invalid configuration must still reject reclamation"
+        );
+        assert!(
+            matches!(wrapper.renew_snapshot_lease(volume, path, token).await, Err(DiskError::FileNotFound)),
+            "failed reclamation must not strand a reader token"
+        );
+        assert_eq!(
+            wrapper.read_all(volume, "object/data/part.1").await.expect("retained data"),
+            Bytes::from_static(b"live shard"),
+            "configuration failure must not fall back to unbudgeted deletion"
+        );
+    }
+
+    #[tokio::test]
+    async fn rpc_cleanup_entries_share_disk_budget() {
+        const CHILD: &str = "RUSTFS_CLEANUP_RPC_TEST_CHILD";
+        if std::env::var_os(CHILD).is_none() {
+            let output = tokio::task::spawn_blocking(|| {
+                std::process::Command::new(std::env::current_exe().unwrap())
+                    .args([
+                        "--exact",
+                        "disk::disk_store::tests::rpc_cleanup_entries_share_disk_budget",
+                        "--nocapture",
+                    ])
+                    .env(CHILD, "1")
+                    .env("RUSTFS_CLEANUP_ISOLATE_ENABLE", "true")
+                    .env("RUSTFS_CLEANUP_ASYNC_THREADS", "1")
+                    .env("RUSTFS_CLEANUP_DISK_MAX_PENDING", "1")
+                    .env("RUSTFS_CLEANUP_DISK_WORKERS", "1")
+                    .env("RUSTFS_CLEANUP_GC_WORKERS", "1")
+                    .env_remove("RUSTFS_CLEANUP_CPUS")
+                    .env_remove("RUSTFS_PUT_RENAME_TAIL_CLEANUP_COUNTERFACTUAL_SKIP")
+                    .env_remove("RUSTFS_PUT_RENAME_TAIL_CLEANUP_DEFER_HOLD_WORKER")
+                    .env_remove("RUSTFS_PUT_RENAME_TAIL_CLEANUP_ZERO_TARGET_TMP_DELETE_SKIP")
+                    .output()
+                    .unwrap()
+            })
+            .await
+            .unwrap();
+            assert!(
+                output.status.success(),
+                "{}\n{}",
+                String::from_utf8_lossy(&output.stdout),
+                String::from_utf8_lossy(&output.stderr)
+            );
+            assert!(String::from_utf8_lossy(&output.stdout).contains("1 passed"));
+            return;
+        }
+        let dir = tempfile::tempdir().unwrap();
+        let endpoint = Endpoint::try_from(dir.path().to_str().unwrap()).unwrap();
+        let disk = Arc::new(LocalDisk::new(&endpoint, false).await.unwrap());
+        let wrapper = LocalDiskWrapper::new(disk, false);
+        let volume = "cleanup-receiver";
+        let path = "object/old-data";
+        let body_path = format!("{path}/part.1");
+        let receipt_path = format!("{path}/{}", cleanup_runtime::OLD_DATA_CLEANUP_RECEIPT_FILE);
+        wrapper.make_volume(volume).await.unwrap();
+        wrapper
+            .write_all(volume, &body_path, Bytes::from_static(b"old body"))
+            .await
+            .unwrap();
+        let read_lease = wrapper
+            .acquire_snapshot_lease(volume, path)
+            .await
+            .expect("ordinary GET lease");
+        let (entered, entry) = tokio::sync::oneshot::channel();
+        let (release, released) = std::sync::mpsc::channel();
+        let occupied = tokio::spawn(cleanup_runtime::disk_operation(async move {
+            entered.send(()).unwrap();
+            // Deliberately stall the only cleanup scheduler to prove that
+            // ordinary GET completion does not dispatch to it.
+            released
+                .recv_timeout(Duration::from_secs(30))
+                .expect("release stalled scheduler");
+            Ok(())
+        }));
+        tokio::time::timeout(Duration::from_secs(20), entry).await.unwrap().unwrap();
+        tokio::time::timeout(Duration::from_secs(5), wrapper.release_snapshot_lease(volume, path, read_lease))
+            .await
+            .expect("ordinary GET release must progress while the cleanup scheduler is stalled")
+            .expect("release without reclamation");
+        let fence_path = super::super::quota_mutation_fence_path(volume, "object");
+        let fence = wrapper
+            .acquire_snapshot_lease(super::super::RUSTFS_META_BUCKET, &fence_path)
+            .await
+            .unwrap();
+        wrapper
+            .release_snapshot_lease(super::super::RUSTFS_META_BUCKET, &fence_path, fence)
+            .await
+            .unwrap();
+        let timeout = wrapper
+            .track_cleanup_mutation(
+                "delete",
+                DiskMetricMutation::Delete,
+                || std::future::pending::<Result<()>>(),
+                Duration::from_millis(10),
+            )
+            .await;
+        assert!(matches!(timeout, Err(DiskError::Timeout)));
+        assert!(!wrapper.health.is_faulty(), "cleanup queue delay must not mark a healthy disk faulty");
+        let options = DeleteOptions {
+            recursive: true,
+            ..Default::default()
+        };
+        let drops = Arc::new(std::sync::atomic::AtomicUsize::new(0));
+        // This is the same entry point used after RPC authentication and
+        // scanner publication-lease validation by handle_delete.
+        let result = wrapper
+            .delete_with_publication_guard(volume, path, options.clone(), Some(Arc::new(DropProbe(drops.clone()))))
+            .await;
+        assert!(matches!(result, Err(DiskError::Io(err)) if err.kind() == io::ErrorKind::WouldBlock));
+        assert_eq!(drops.load(Ordering::SeqCst), 1);
+        assert!(matches!(wrapper.delete_data_dir(volume, path, options.clone()).await,
+            Err(DiskError::Io(err)) if err.kind() == io::ErrorKind::WouldBlock));
+        assert!(matches!(wrapper.write_all(volume, &receipt_path, Bytes::from_static(b"receipt")).await,
+            Err(DiskError::Io(err)) if err.kind() == io::ErrorKind::WouldBlock));
+        assert_eq!(wrapper.read_all(volume, &body_path).await.unwrap(), Bytes::from_static(b"old body"));
+        release.send(()).unwrap();
+        occupied.await.unwrap().unwrap();
+        wrapper
+            .write_all(volume, &receipt_path, Bytes::from_static(b"receipt"))
+            .await
+            .unwrap();
+        wrapper
+            .delete_with_publication_guard(volume, path, options, None)
+            .await
+            .unwrap();
+        assert!(matches!(wrapper.read_all(volume, &body_path).await, Err(DiskError::FileNotFound)));
+        // Snapshot readers defer old-data cleanup. Saturation must release
+        // their token while retaining reclaim intent, rather than leaking a
+        // live token or performing an unbudgeted delete on the RPC runtime.
+        wrapper
+            .write_all(volume, &body_path, Bytes::from_static(b"leased body"))
+            .await
+            .unwrap();
+        let lease = wrapper.acquire_snapshot_lease(volume, path).await.unwrap();
+        assert_eq!(
+            wrapper
+                .delete_data_dir(
+                    volume,
+                    path,
+                    DeleteOptions {
+                        recursive: true,
+                        ..Default::default()
+                    }
+                )
+                .await
+                .unwrap(),
+            DataDirDeleteStatus::Deferred
+        );
+        let (entered, entry) = tokio::sync::oneshot::channel();
+        let (release, released) = tokio::sync::oneshot::channel();
+        let occupied = tokio::spawn(cleanup_runtime::disk_operation(async move {
+            entered.send(()).unwrap();
+            released.await.unwrap();
+            Ok(())
+        }));
+        tokio::time::timeout(Duration::from_secs(20), entry).await.unwrap().unwrap();
+        assert!(matches!(wrapper.release_snapshot_lease(volume, path, lease).await,
+            Err(DiskError::Io(err)) if err.kind() == io::ErrorKind::WouldBlock));
+        assert_eq!(wrapper.read_all(volume, &body_path).await.unwrap(), Bytes::from_static(b"leased body"));
+        release.send(()).unwrap();
+        occupied.await.unwrap().unwrap();
+        assert_eq!(
+            wrapper
+                .delete_data_dir(
+                    volume,
+                    path,
+                    DeleteOptions {
+                        recursive: true,
+                        ..Default::default()
+                    }
+                )
+                .await
+                .unwrap(),
+            DataDirDeleteStatus::Deleted
+        );
+        assert!(matches!(wrapper.read_all(volume, &body_path).await, Err(DiskError::FileNotFound)));
+    }
+
     #[tokio::test]
     async fn owned_mutation_keeps_publication_guard_after_waiter_cancellation() {
         let drops = Arc::new(std::sync::atomic::AtomicUsize::new(0));

crates/ecstore/src/disk/local.rs

@@ -5093,7 +5093,7 @@ pub struct LocalDisk {
     /// pathname is later covered by another mount.
     io_root: PathBuf,
     #[cfg(target_os = "linux")]
-    mount_lease: std::fs::File,
+    mount_lease: Arc<std::fs::File>,
     #[cfg(target_os = "linux")]
     mount_lease_mount_id: Option<u64>,
     /// Public path for callers that need the configured disk layout. Internal
@@ -5551,7 +5551,7 @@ impl LocalDisk {
             publication_root,
             io_root: io_root.clone(),
             #[cfg(target_os = "linux")]
-            mount_lease,
+            mount_lease: Arc::new(mount_lease),
             #[cfg(target_os = "linux")]
             mount_lease_mount_id,
             endpoint: ep.clone(),
@@ -5606,7 +5606,13 @@ impl LocalDisk {
 
         let io_root = disk.io_root.clone();
         let publication_root = disk.publication_root.clone();
-        tokio::spawn(Self::cleanup_deleted_objects_loop(io_root, publication_root, exit_rx));
+        tokio::spawn(Self::cleanup_deleted_objects_loop(
+            io_root,
+            publication_root,
+            #[cfg(target_os = "linux")]
+            disk.mount_lease.clone(),
+            exit_rx,
+        ));
         debug!(
             event = EVENT_DISK_LOCAL_STARTUP_CLEANUP,
             component = LOG_COMPONENT_ECSTORE,
@@ -5622,30 +5628,53 @@ impl LocalDisk {
     async fn cleanup_deleted_objects_loop(
         root: PathBuf,
         publication_root: os::PublicationRoot,
+        #[cfg(target_os = "linux")] mount_lease: Arc<std::fs::File>,
         mut exit_rx: tokio::sync::broadcast::Receiver<()>,
     ) {
         let start_at = Instant::now() + DELETED_OBJECTS_CLEANUP_INTERVAL;
         let mut interval = interval_at(start_at, DELETED_OBJECTS_CLEANUP_INTERVAL);
+        interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
         loop {
             tokio::select! {
                 _ = interval.tick() => {
-                    if let Err(err) = Self::cleanup_deleted_objects(root.clone()).await {
+                    let root = root.clone();
+                    let publication_root = publication_root.clone();
+                    #[cfg(target_os = "linux")]
+                    let mount_lease = mount_lease.clone();
+                    if let Err(err) = super::cleanup_runtime::run_gc(move |budget| async move {
+                        // Queued/detached scans must keep /proc/self/fd paths
+                        // attached to the original mount after LocalDisk drops.
+                        #[cfg(target_os = "linux")]
+                        let _mount_lease = mount_lease;
+                        if let Err(err) = Self::cleanup_deleted_objects(root.clone(), &budget).await {
+                            error!(
+                                event = EVENT_DISK_LOCAL_BACKGROUND_CLEANUP,
+                                component = LOG_COMPONENT_ECSTORE,
+                                subsystem = LOG_SUBSYSTEM_DISK_LOCAL,
+                                task = "deleted_objects",
+                                state = "failed",
+                                error = ?err,
+                                "Disk local background cleanup failed"
+                            );
+                        }
+                        if let Err(err) = Self::cleanup_stale_tmp_objects(root.clone(), &publication_root, &budget).await {
+                            error!(
+                                event = EVENT_DISK_LOCAL_BACKGROUND_CLEANUP,
+                                component = LOG_COMPONENT_ECSTORE,
+                                subsystem = LOG_SUBSYSTEM_DISK_LOCAL,
+                                task = "stale_tmp_objects",
+                                state = "failed",
+                                error = ?err,
+                                "Disk local background cleanup failed"
+                            );
+                        }
+                        Ok(())
+                    }).await {
                         error!(
                             event = EVENT_DISK_LOCAL_BACKGROUND_CLEANUP,
                             component = LOG_COMPONENT_ECSTORE,
                             subsystem = LOG_SUBSYSTEM_DISK_LOCAL,
-                            task = "deleted_objects",
-                            state = "failed",
-                            error = ?err,
-                            "Disk local background cleanup failed"
-                        );
-                    }
-                    if let Err(err) = Self::cleanup_stale_tmp_objects(root.clone(), &publication_root).await {
-                        error!(
-                            event = EVENT_DISK_LOCAL_BACKGROUND_CLEANUP,
-                            component = LOG_COMPONENT_ECSTORE,
-                            subsystem = LOG_SUBSYSTEM_DISK_LOCAL,
-                            task = "stale_tmp_objects",
+                            task = "cleanup_dispatch",
                             state = "failed",
                             error = ?err,
                             "Disk local background cleanup failed"
@@ -5736,95 +5765,123 @@ impl LocalDisk {
         }
     }
 
-    async fn cleanup_stale_tmp_objects(root: PathBuf, publication_root: &os::PublicationRoot) -> Result<()> {
-        Self::cleanup_stale_tmp_objects_with_expiry(root, publication_root, STALE_TMP_OBJECT_EXPIRY).await
+    async fn cleanup_stale_tmp_objects(
+        root: PathBuf,
+        publication_root: &os::PublicationRoot,
+        budget: &super::cleanup_runtime::GcBudget,
+    ) -> Result<()> {
+        Self::cleanup_stale_tmp_objects_with_expiry(root, publication_root, STALE_TMP_OBJECT_EXPIRY, budget).await
     }
 
     async fn cleanup_stale_tmp_objects_with_expiry(
         root: PathBuf,
         publication_root: &os::PublicationRoot,
         expiry: Duration,
+        budget: &super::cleanup_runtime::GcBudget,
     ) -> Result<()> {
         let tmp_path = Self::meta_path(&root, RUSTFS_META_TMP_BUCKET);
-        let mut entries = match fs::read_dir(&tmp_path).await {
+        let mut entries = match budget.step(async { Ok(fs::read_dir(&tmp_path).await?) }).await {
             Ok(entries) => entries,
             Err(e) => {
-                if e.kind() == ErrorKind::NotFound {
+                if matches!(&e, DiskError::Io(err) if err.kind() == ErrorKind::NotFound) {
                     return Ok(());
                 }
-                return Err(e.into());
+                return Err(e);
             }
         };
 
-        while let Some(entry) = entries.next_entry().await? {
-            let name = entry.file_name().to_string_lossy().to_string();
-            if name.is_empty() || name == "." || name == ".." || name == ".trash" {
-                continue;
+        loop {
+            let more = budget
+                .step(async {
+                    let Some(entry) = entries.next_entry().await? else {
+                        return Ok(false);
+                    };
+                    Self::cleanup_stale_tmp_entry(entry, &root, publication_root, expiry).await?;
+                    Ok(true)
+                })
+                .await?;
+            if !more {
+                break;
             }
-
-            let file_type = entry.file_type().await?;
-            if !file_type.is_dir() {
-                continue;
-            }
-
-            let Some(age) = entry
-                .metadata()
-                .await?
-                .modified()
-                .ok()
-                .and_then(|modified| modified.elapsed().ok())
-            else {
-                continue;
-            };
-            if age <= expiry {
-                continue;
-            }
-
-            let target_path = Self::meta_path(&root, RUSTFS_META_TMP_DELETED_BUCKET).join(Uuid::new_v4().to_string());
-            rename_all(entry.path(), target_path, Self::meta_path(&root, RUSTFS_META_BUCKET), publication_root).await?;
         }
 
         Ok(())
     }
 
-    async fn cleanup_deleted_objects(root: PathBuf) -> Result<()> {
+    async fn cleanup_stale_tmp_entry(
+        entry: fs::DirEntry,
+        root: &Path,
+        publication_root: &os::PublicationRoot,
+        expiry: Duration,
+    ) -> Result<()> {
+        let name = entry.file_name().to_string_lossy().to_string();
+        if name.is_empty() || name == "." || name == ".." || name == ".trash" || !entry.file_type().await?.is_dir() {
+            return Ok(());
+        }
+        let Some(age) = entry
+            .metadata()
+            .await?
+            .modified()
+            .ok()
+            .and_then(|modified| modified.elapsed().ok())
+        else {
+            return Ok(());
+        };
+        if age <= expiry {
+            return Ok(());
+        }
+        let target_path = Self::meta_path(root, RUSTFS_META_TMP_DELETED_BUCKET).join(Uuid::new_v4().to_string());
+        rename_all(entry.path(), target_path, Self::meta_path(root, RUSTFS_META_BUCKET), publication_root).await?;
+        Ok(())
+    }
+
+    async fn cleanup_deleted_objects(root: PathBuf, budget: &super::cleanup_runtime::GcBudget) -> Result<()> {
         let trash = Self::meta_path(&root, RUSTFS_META_TMP_DELETED_BUCKET);
-        let mut entries = match fs::read_dir(&trash).await {
+        let mut entries = match budget.step(async { Ok(fs::read_dir(&trash).await?) }).await {
             Ok(entries) => entries,
             Err(e) => {
-                if e.kind() == ErrorKind::NotFound {
+                if matches!(&e, DiskError::Io(err) if err.kind() == ErrorKind::NotFound) {
                     return Ok(());
                 }
-                return Err(e.into());
+                return Err(e);
             }
         };
 
-        while let Some(entry) = entries.next_entry().await? {
-            let name = entry.file_name().to_string_lossy().to_string();
-            if name.is_empty() || name == "." || name == ".." {
-                continue;
-            }
-
-            let file_type = entry.file_type().await?;
-
-            let path = trash.join(name);
-
-            if file_type.is_dir() {
-                if let Err(e) = tokio::fs::remove_dir_all(path).await
-                    && e.kind() != ErrorKind::NotFound
-                {
-                    return Err(e.into());
-                }
-            } else if let Err(e) = tokio::fs::remove_file(path).await
-                && e.kind() != ErrorKind::NotFound
-            {
-                return Err(e.into());
+        loop {
+            let more = budget
+                .step(async {
+                    let Some(entry) = entries.next_entry().await? else {
+                        return Ok(false);
+                    };
+                    Self::cleanup_trash_entry(entry, &trash).await?;
+                    Ok(true)
+                })
+                .await?;
+            if !more {
+                break;
             }
         }
 
         Ok(())
     }
 
+    async fn cleanup_trash_entry(entry: fs::DirEntry, trash: &Path) -> Result<()> {
+        let name = entry.file_name().to_string_lossy().to_string();
+        if name.is_empty() || name == "." || name == ".." {
+            return Ok(());
+        }
+        let path = trash.join(name);
+        let result = if entry.file_type().await?.is_dir() {
+            fs::remove_dir_all(path).await
+        } else {
+            fs::remove_file(path).await
+        };
+        match result {
+            Err(err) if err.kind() != ErrorKind::NotFound => Err(err.into()),
+            _ => Ok(()),
+        }
+    }
+
     fn is_valid_volname(volname: &str) -> bool {
         if volname.len() < 3 {
             return false;
@@ -6135,6 +6192,36 @@ impl LocalDisk {
         Ok(())
     }
 
+    /// Keep the common GET completion on its caller. Only a last-reader
+    /// release with pending reclamation needs an owned cleanup task. Leave
+    /// that token untouched so the slow path owns the complete transition.
+    pub(in crate::disk) async fn release_snapshot_lease_without_cleanup(
+        &self,
+        volume: &str,
+        path: &str,
+        token: SnapshotLeaseToken,
+    ) -> bool {
+        let key = SnapshotLeaseKey {
+            volume: volume.to_owned(),
+            path: path.to_owned(),
+        };
+        let mut registry = self.snapshot_leases.lock().await;
+        let Some(entry) = registry.entries.get_mut(&key) else {
+            return true;
+        };
+        if !entry.deleting
+            && entry.pending_delete.is_some()
+            && (entry.tokens.is_empty() || (entry.tokens.len() == 1 && entry.tokens.contains(&token)))
+        {
+            return false;
+        }
+        entry.tokens.remove(&token);
+        if entry.tokens.is_empty() && !entry.deleting {
+            registry.entries.remove(&key);
+        }
+        true
+    }
+
     async fn delete_data_dir_with_namespace_owner(
         &self,
         volume: &str,
@@ -10723,7 +10810,16 @@ impl DiskAPI for LocalDisk {
             entry.deleting = true;
             opts
         };
-        let result = self.delete_unleased(volume, path, &opts).await;
+        // Release the reader token even under saturation. A rejected reclaim
+        // retains pending_delete (and its on-disk receipt) for a later retry.
+        // Do not wait for execution while a releasing reader may own locks.
+        let result = match super::cleanup_runtime::try_disk_execution() {
+            Ok(permit) => {
+                let _permit = permit;
+                self.delete_unleased(volume, path, &opts).await
+            }
+            Err(err) => Err(err.into()),
+        };
         let mut registry = self.snapshot_leases.lock().await;
         match result {
             Ok(()) => {
@@ -12509,10 +12605,15 @@ mod test {
         disk.wait_for_startup_cleanup().await;
         assert_eq!(disk.startup_cleanup_ready.load(Ordering::Acquire), 1);
 
-        LocalDisk::cleanup_stale_tmp_objects_with_expiry(dir.path().join("missing-root"), &publication_root, Duration::ZERO)
-            .await
-            .expect("missing tmp path should be a cleanup no-op");
-        LocalDisk::cleanup_deleted_objects(dir.path().join("missing-root"))
+        LocalDisk::cleanup_stale_tmp_objects_with_expiry(
+            dir.path().join("missing-root"),
+            &publication_root,
+            Duration::ZERO,
+            &crate::disk::cleanup_runtime::GcBudget::default(),
+        )
+        .await
+        .expect("missing tmp path should be a cleanup no-op");
+        LocalDisk::cleanup_deleted_objects(dir.path().join("missing-root"), &crate::disk::cleanup_runtime::GcBudget::default())
             .await
             .expect("missing trash path should be a cleanup no-op");
 
@@ -12525,9 +12626,14 @@ mod test {
         fs::create_dir_all(&trash_root).await.expect("trash dir should be created");
         backdate_mtime(&stale_dir, Duration::from_secs(10));
 
-        LocalDisk::cleanup_stale_tmp_objects_with_expiry(dir.path().to_path_buf(), &publication_root, Duration::ZERO)
-            .await
-            .expect("stale tmp directory should move to trash");
+        LocalDisk::cleanup_stale_tmp_objects_with_expiry(
+            dir.path().to_path_buf(),
+            &publication_root,
+            Duration::ZERO,
+            &crate::disk::cleanup_runtime::GcBudget::default(),
+        )
+        .await
+        .expect("stale tmp directory should move to trash");
         assert!(!stale_dir.exists(), "stale tmp directory should be moved away");
         assert!(live_file.exists(), "plain tmp files should be ignored by stale dir cleanup");
 
@@ -12537,7 +12643,7 @@ mod test {
         fs::create_dir_all(trash_root.join("trash-dir"))
             .await
             .expect("trash dir should be created");
-        LocalDisk::cleanup_deleted_objects(dir.path().to_path_buf())
+        LocalDisk::cleanup_deleted_objects(dir.path().to_path_buf(), &crate::disk::cleanup_runtime::GcBudget::default())
             .await
             .expect("trash cleanup should remove files and directories");
         assert!(
@@ -18199,9 +18305,14 @@ mod test {
         // Backdate after the write above: creating stale/data refreshes the
         // scanned tmp/stale directory's mtime.
         backdate_mtime(&tmp.join("stale"), Duration::from_secs(10));
-        LocalDisk::cleanup_stale_tmp_objects_with_expiry(dir.path().to_path_buf(), &publication_root, Duration::ZERO)
-            .await
-            .expect("operation should succeed");
+        LocalDisk::cleanup_stale_tmp_objects_with_expiry(
+            dir.path().to_path_buf(),
+            &publication_root,
+            Duration::ZERO,
+            &crate::disk::cleanup_runtime::GcBudget::default(),
+        )
+        .await
+        .expect("operation should succeed");
 
         assert!(!tmp.join("stale").exists());
         assert!(trash.exists());
@@ -18228,9 +18339,14 @@ mod test {
         fs::write(&fresh_dir, b"temporary").await.expect("operation should succeed");
         fs::write(&regular_file, b"keep").await.expect("operation should succeed");
 
-        LocalDisk::cleanup_stale_tmp_objects_with_expiry(dir.path().to_path_buf(), &publication_root, Duration::from_secs(60))
-            .await
-            .expect("operation should succeed");
+        LocalDisk::cleanup_stale_tmp_objects_with_expiry(
+            dir.path().to_path_buf(),
+            &publication_root,
+            Duration::from_secs(60),
+            &crate::disk::cleanup_runtime::GcBudget::default(),
+        )
+        .await
+        .expect("operation should succeed");
 
         assert!(tmp.join("fresh").exists());
         assert!(regular_file.exists());
@@ -18239,6 +18355,44 @@ mod test {
         assert!(entries.next_entry().await.expect("operation should succeed").is_none());
     }
 
+    #[cfg(target_os = "linux")]
+    #[tokio::test]
+    async fn background_cleanup_keeps_mount_lease_after_disk_drop() {
+        let dir = tempfile::tempdir().unwrap();
+        let endpoint = Endpoint::try_from(dir.path().to_str().unwrap()).unwrap();
+        let disk = LocalDisk::new(&endpoint, false).await.unwrap();
+        let root = disk.io_root.clone();
+        let trash_entry = LocalDisk::meta_path(&root, RUSTFS_META_TMP_DELETED_BUCKET).join("detached-gc");
+        fs::create_dir_all(&trash_entry).await.unwrap();
+        fs::write(trash_entry.join("part.1"), b"garbage").await.unwrap();
+        let live_file = root.join("keep");
+        fs::write(&live_file, b"live").await.unwrap();
+        let (exit, exited) = tokio::sync::broadcast::channel(1);
+        tokio::time::pause();
+        let cleanup = tokio::spawn(LocalDisk::cleanup_deleted_objects_loop(
+            root.clone(),
+            disk.publication_root.clone(),
+            disk.mount_lease.clone(),
+            exited,
+        ));
+        tokio::task::yield_now().await;
+        drop(disk);
+        tokio::task::yield_now().await;
+        assert!(live_file.exists(), "the GC owner must retain the original /proc/self/fd root");
+        tokio::time::advance(DELETED_OBJECTS_CLEANUP_INTERVAL + Duration::from_secs(1)).await;
+        tokio::time::resume();
+        timeout(Duration::from_secs(20), async {
+            while trash_entry.exists() {
+                tokio::time::sleep(Duration::from_millis(5)).await;
+            }
+        })
+        .await
+        .expect("detached cleanup must finish against the pinned mount");
+        assert_eq!(fs::read(&live_file).await.unwrap(), b"live");
+        exit.send(()).unwrap();
+        timeout(Duration::from_secs(20), cleanup).await.unwrap().unwrap();
+    }
+
     #[tokio::test(start_paused = true)]
     async fn cleanup_loop_interval_does_not_tick_immediately() {
         let start_at = Instant::now() + DELETED_OBJECTS_CLEANUP_INTERVAL;
@@ -21869,9 +22023,14 @@ mod test {
             Bytes::from_static(b"later")
         );
 
-        disk.release_snapshot_lease(volume, &data_dir, renewed)
-            .await
-            .expect("renewed lease release should succeed");
+        assert!(
+            disk.release_snapshot_lease_without_cleanup(volume, &data_dir, renewed).await,
+            "a non-final reader release must not dispatch cleanup"
+        );
+        assert!(
+            !disk.release_snapshot_lease_without_cleanup(volume, &data_dir, second).await,
+            "the final reader must retain its token for owned deferred reclamation"
+        );
         assert!(
             disk.read_all(volume, &first_part).await.is_ok(),
             "one remaining lease must keep the data directory"

crates/ecstore/src/disk/mod.rs

@@ -14,6 +14,7 @@
 
 // #730: disk abstractions still carry staged health and direct-I/O migration paths.
 
+pub(crate) mod cleanup_runtime;
 pub mod disk_store;
 pub mod endpoint;
 pub mod error;

crates/ecstore/src/disk/os.rs

@@ -670,6 +670,8 @@ fn file_fdatasync_group_commit_wait() -> Duration {
 #[derive(Clone, Eq, Hash, PartialEq)]
 struct DstDirFsyncGroupKey {
     canonical_path: PathBuf,
+    // A batch must execute in the same resource class as its producers.
+    isolated: bool,
     #[cfg(unix)]
     dev: u64,
     #[cfg(unix)]
@@ -686,13 +688,17 @@ impl DstDirFsyncGroupKey {
             use std::os::unix::fs::MetadataExt;
             Ok(Self {
                 canonical_path,
+                isolated: FSYNC_ON_CURRENT_RUNTIME.get(),
                 dev: metadata.dev(),
                 ino: metadata.ino(),
             })
         }
         #[cfg(not(unix))]
         {
-            Ok(Self { canonical_path })
+            Ok(Self {
+                canonical_path,
+                isolated: FSYNC_ON_CURRENT_RUNTIME.get(),
+            })
         }
     }
 }
@@ -1061,7 +1067,7 @@ fn clear_dst_dir_fsync_group_commit_for_test() {
     DST_DIR_FSYNC_GROUP_COMMIT.clear_for_test();
 }
 
-type FileFdatasyncGroupKey = usize;
+type FileFdatasyncGroupKey = (usize, bool);
 
 struct FileFdatasyncWaiter {
     files: Vec<PathBuf>,
@@ -1133,7 +1139,7 @@ impl FileFdatasyncGroupCommit {
             ));
         }
         let (result_tx, result_rx) = oneshot::channel();
-        let key = Arc::as_ptr(&disk_permits) as FileFdatasyncGroupKey;
+        let key = (Arc::as_ptr(&disk_permits) as usize, FSYNC_ON_CURRENT_RUNTIME.get());
         let mut registry = self.inner.lock();
         registry.groups.retain(|_, group| group.disk_permits.strong_count() > 0);
         if registry.total_waiters >= MAX_FILE_FDATASYNC_WAITERS {
@@ -1371,9 +1377,26 @@ static FSYNC_RUNTIME: LazyLock<Option<tokio::runtime::Runtime>> = LazyLock::new(
     }
 });
 
+thread_local! {
+    static FSYNC_ON_CURRENT_RUNTIME: std::cell::Cell<bool> = const { std::cell::Cell::new(false) };
+}
+
+/// Called only when starting threads belonging to an isolated I/O runtime.
+/// Async and blocking threads use the same runtime's blocking budget for fsync.
+pub(crate) fn use_current_runtime_for_fsync() {
+    FSYNC_ON_CURRENT_RUNTIME.set(true);
+}
+
+pub(super) fn is_isolated_io_thread() -> bool {
+    FSYNC_ON_CURRENT_RUNTIME.get()
+}
+
 /// Spawn a blocking task on the fsync-dedicated runtime if configured,
 /// otherwise fall back to the main tokio blocking pool.
 fn fsync_spawn_blocking<T: Send + 'static>(f: impl FnOnce() -> T + Send + 'static) -> tokio::task::JoinHandle<T> {
+    if FSYNC_ON_CURRENT_RUNTIME.get() {
+        return tokio::task::spawn_blocking(f);
+    }
     match FSYNC_RUNTIME.as_ref() {
         Some(rt) => rt.spawn_blocking(f),
         None => tokio::task::spawn_blocking(f),
@@ -4484,6 +4507,60 @@ mod tests {
     use tempfile::tempdir;
     use tracing_subscriber::fmt::MakeWriter;
 
+    #[test]
+    fn fsync_groups_do_not_mix_cleanup_and_foreground_workers() {
+        let dir = tempdir().unwrap();
+        let dst_groups = Arc::new(DstDirFsyncGroupCommit::default());
+        let file_groups = Arc::new(FileFdatasyncGroupCommit::default());
+        let disk_permits = Arc::new(Semaphore::new(1));
+        let (_, dst_worker) = dst_groups.enqueue_for_test(dir.path()).unwrap();
+        let (_, file_worker) = file_groups
+            .enqueue(disk_permits.clone(), vec![dir.path().join("one")])
+            .unwrap();
+        assert!(dst_worker.is_some());
+        assert!(file_worker.is_some());
+        let cleanup_dst_groups = dst_groups.clone();
+        let cleanup_file_groups = file_groups.clone();
+        std::thread::spawn(move || {
+            use_current_runtime_for_fsync();
+            let (_, dst_worker) = cleanup_dst_groups.enqueue_for_test(dir.path()).unwrap();
+            let (_, file_worker) = cleanup_file_groups
+                .enqueue(disk_permits, vec![dir.path().join("two")])
+                .unwrap();
+            assert!(dst_worker.is_some(), "cleanup must not join a foreground directory worker");
+            assert!(file_worker.is_some(), "cleanup must not join a foreground file worker");
+        })
+        .join()
+        .unwrap();
+        assert_eq!(dst_groups.inner.lock().groups.len(), 2);
+        assert_eq!(file_groups.inner.lock().groups.len(), 2);
+    }
+
+    #[tokio::test]
+    async fn isolated_runtime_keeps_fsync_on_its_own_blocking_pool() {
+        let runtime = tokio::runtime::Builder::new_multi_thread()
+            .worker_threads(1)
+            .max_blocking_threads(1)
+            .thread_name("isolated-fsync-test")
+            .on_thread_start(use_current_runtime_for_fsync)
+            .enable_all()
+            .build()
+            .unwrap();
+        let result = runtime
+            .spawn(async {
+                fsync_spawn_blocking(|| {
+                    assert_eq!(std::thread::current().name(), Some("isolated-fsync-test"));
+                })
+                .await
+                .unwrap();
+            })
+            .await;
+        // A runtime cannot be dropped from another runtime's async context.
+        runtime.shutdown_background();
+        result.unwrap();
+        assert!(!FSYNC_ON_CURRENT_RUNTIME.get(), "the calling runtime must retain normal fsync routing");
+    }
+
     fn file_sync_limiter() -> Arc<Semaphore> {
         Arc::new(Semaphore::new(MAX_PARALLEL_FILE_SYNCS))
     }

crates/ecstore/src/set_disk/ops/object.rs

@@ -20,6 +20,7 @@
 //! SetDisks core (io_primitives) via inherent calls.
 
 use crate::core::pools::DecommissionCapacityAdmission;
+use crate::disk::cleanup_runtime::{self, OLD_DATA_CLEANUP_RECEIPT_FILE};
 use rustfs_common::mrf_channel::MrfDeleteMarkerPurge;
 use rustfs_filemeta::metadata_keys;
 
@@ -885,7 +886,6 @@ async fn acquire_single_tier_delete_lease(opts: &ObjectOptions, source: &ObjectI
     Ok(Some(lease))
 }
 
-const OLD_DATA_CLEANUP_RECEIPT_FILE: &str = ".rustfs-old-data-cleanup-receipt.json";
 const SCANNER_PUBLICATION_LEASE_FENCE_MAX_BYTES: usize = 64 * 1024;
 const SCANNER_PUBLICATION_LEASE_FENCE_MAX_ENTRIES: usize = 256;
 
@@ -3686,6 +3686,19 @@ pub(in crate::set_disk) fn merge_replication_metadata_lww(
 }
 
 impl SetDisks {
+    /// Failure cleanup stays awaited, but shares the same execution budget as
+    /// successful PUT continuations when isolation is enabled.
+    async fn delete_put_tmp_with_budget(&self, tmp_dir: &str, reservation: &Option<cleanup_runtime::Reservation>) -> Result<()> {
+        if reservation.is_none() {
+            return self.delete_all(RUSTFS_META_TMP_BUCKET, tmp_dir).await;
+        }
+        let set = self.clone();
+        let tmp_dir = tmp_dir.to_owned();
+        cleanup_runtime::spawn(reservation.clone(), async move { set.delete_all(RUSTFS_META_TMP_BUCKET, &tmp_dir).await })
+            .await
+            .map_err(Error::other)?
+    }
+
     pub(in crate::set_disk) async fn cleanup_rename_tail(
         &self,
         targets: Vec<RenameTailCleanup>,
@@ -3981,6 +3994,23 @@ impl SetDisks {
             .scanner_publication_commit_scope
             .clone()
             .map(ScannerPublicationCommitScopeGuard::new);
+        // Reserve before staging. Some callers already hold external locks,
+        // so admission must not wait for another PUT to release its budget.
+        // Commit/cleanup owners retain clones across cancellation and early ACK.
+        if opts
+            .put_object_cancellation
+            .as_ref()
+            .is_some_and(CancellationToken::is_cancelled)
+        {
+            return Err(StorageError::OperationCanceled);
+        }
+        let cleanup_reservation = cleanup_runtime::reserve().map_err(|err| {
+            if err.kind() == std::io::ErrorKind::WouldBlock {
+                StorageError::SlowDown
+            } else {
+                Error::other(err)
+            }
+        })?;
         let storage_class_config = self.storage_class_config_snapshot();
         self.invalidate_get_object_metadata_cache(bucket, object).await;
 
@@ -4931,6 +4961,7 @@ impl SetDisks {
             let commit_tier_free_version_receipt_sink = opts.tier_free_version_receipt_sink.clone();
             let commit_skip_free_version = opts.skip_free_version;
             let request_cancellation = operation_cancellation.clone();
+            let commit_cleanup_reservation = cleanup_reservation.clone();
             tmp_cleanup_owned = true;
             rustfs_io_metrics::record_put_object_stage_duration_from(
                 rustfs_io_metrics::PUT_STAGE_PUT_OBJECT_COMMIT_READY_CONTEXT_PREPARE,
@@ -5129,7 +5160,10 @@ impl SetDisks {
                         quota_mutation_fence,
                     )
                     .await;
-                    if let Err(cleanup_err) = commit_set.delete_all(RUSTFS_META_TMP_BUCKET, &commit_tmp_dir).await {
+                    if let Err(cleanup_err) = commit_set
+                        .delete_put_tmp_with_budget(&commit_tmp_dir, &commit_cleanup_reservation)
+                        .await
+                    {
                         warn!(tmp_dir = %commit_tmp_dir, error = ?cleanup_err, "failed to cleanup put_object temporary data");
                     } else if issue3031_diag_enabled() {
                         warn!(
@@ -5287,6 +5321,7 @@ impl SetDisks {
                         let fence_object = commit_object.clone();
                         let (guard_release_tx, guard_release_rx) = tokio::sync::oneshot::channel();
                         rename_guard_release = Some(guard_release_tx);
+                        let cleanup_reservation = commit_cleanup_reservation.clone();
                         tokio::spawn(finish_rename_tail_heal(
                             rename_tail_drain,
                             guard_release_rx,
@@ -5321,7 +5356,7 @@ impl SetDisks {
                                 drop(object_lock_guard);
                                 drop(publication_guard);
                                 drop(bucket_lifecycle_guard);
-                                async move {
+                                cleanup_runtime::run(cleanup_reservation, async move {
                                     let zero_target_cleanup = targets.is_empty();
                                     rustfs_io_metrics::record_put_object_stage_duration(
                                         rustfs_io_metrics::PUT_STAGE_SET_DISK_RENAME_TAIL_CLEANUP_TARGET_COUNT,
@@ -5370,7 +5405,7 @@ impl SetDisks {
                                         rustfs_io_metrics::PUT_STAGE_SET_DISK_RENAME_TAIL_CLEANUP_DECOMMISSION_GUARD_DROP,
                                         guard_drop_started,
                                     );
-                                }
+                                })
                             },
                             |request| async move { heal_set.submit_rename_tail_heal(request).await },
                         ));
@@ -5427,7 +5462,10 @@ impl SetDisks {
                             // rollback backup for recovery; cleanup is unsafe.
                             return Err(err.into());
                         }
-                        if let Err(cleanup_err) = commit_set.delete_all(RUSTFS_META_TMP_BUCKET, &commit_tmp_dir).await {
+                        if let Err(cleanup_err) = commit_set
+                            .delete_put_tmp_with_budget(&commit_tmp_dir, &commit_cleanup_reservation)
+                            .await
+                        {
                             warn!(tmp_dir = %commit_tmp_dir, error = ?cleanup_err, "failed to cleanup put_object temporary data");
                         } else if issue3031_diag_enabled() {
                             warn!(
@@ -5508,16 +5546,36 @@ impl SetDisks {
                         1.0,
                     );
                     let cleanup_receipt_started = rustfs_io_metrics::put_stage_timer();
-                    commit_set
-                        .persist_old_data_cleanup_receipts(
-                            &cleanup_disks,
-                            &commit_bucket,
-                            &commit_object,
-                            old_dir,
-                            committed_data_dir,
-                            transaction_epoch,
-                        )
+                    if commit_cleanup_reservation.is_none() {
+                        commit_set
+                            .persist_old_data_cleanup_receipts(
+                                &cleanup_disks,
+                                &commit_bucket,
+                                &commit_object,
+                                old_dir,
+                                committed_data_dir,
+                                transaction_epoch,
+                            )
+                            .await;
+                    } else {
+                        let cleanup_set = commit_set.clone();
+                        let receipt_disks = cleanup_disks.clone();
+                        let receipt_bucket = commit_bucket.clone();
+                        let receipt_object = commit_object.clone();
+                        cleanup_runtime::run(commit_cleanup_reservation.clone(), async move {
+                            cleanup_set
+                                .persist_old_data_cleanup_receipts(
+                                    &receipt_disks,
+                                    &receipt_bucket,
+                                    &receipt_object,
+                                    old_dir,
+                                    committed_data_dir,
+                                    transaction_epoch,
+                                )
+                                .await;
+                        })
                         .await;
+                    }
                     rustfs_io_metrics::record_put_object_stage_duration_from(
                         rustfs_io_metrics::PUT_STAGE_PUT_OBJECT_COMMIT_NAMESPACE_LOCK_HELD_CLEANUP_RECEIPT,
                         cleanup_receipt_started,
@@ -5582,16 +5640,41 @@ impl SetDisks {
                     // here must not negate an already-committed, durable write, so we
                     // deliberately do NOT `?`-propagate it into a 503. On residue the
                     // report path emits the leak metric and enqueues a heal.
-                    let cleanup = commit_set
-                        .commit_rename_data_dir_and_mark_capacity(
-                            &cleanup_disks,
-                            &commit_bucket,
-                            &commit_object,
-                            &old_dir.to_string(),
-                            &committed_dir,
-                            write_quorum,
-                        )
+                    if commit_cleanup_reservation.is_none() {
+                        let cleanup = commit_set
+                            .commit_rename_data_dir_and_mark_capacity(
+                                &cleanup_disks,
+                                &commit_bucket,
+                                &commit_object,
+                                &old_dir.to_string(),
+                                &committed_dir,
+                                write_quorum,
+                            )
+                            .await;
+                        commit_set
+                            .report_old_data_dir_cleanup(&commit_bucket, &commit_object, &old_dir.to_string(), &cleanup)
+                            .await;
+                    } else {
+                        let cleanup_set = commit_set.clone();
+                        let cleanup_bucket = commit_bucket.clone();
+                        let cleanup_object = commit_object.clone();
+                        cleanup_runtime::run(commit_cleanup_reservation.clone(), async move {
+                            let cleanup = cleanup_set
+                                .commit_rename_data_dir_and_mark_capacity(
+                                    &cleanup_disks,
+                                    &cleanup_bucket,
+                                    &cleanup_object,
+                                    &old_dir.to_string(),
+                                    &committed_dir,
+                                    write_quorum,
+                                )
+                                .await;
+                            cleanup_set
+                                .report_old_data_dir_cleanup(&cleanup_bucket, &cleanup_object, &old_dir.to_string(), &cleanup)
+                                .await;
+                        })
                         .await;
+                    }
                     let cleanup_elapsed = cleanup_stage_start.elapsed();
                     let cleanup_ms = cleanup_elapsed.as_millis() as u64;
                     cleanup_stage_ms = Some(cleanup_ms);
@@ -5599,9 +5682,6 @@ impl SetDisks {
                         "set_disk_old_data_cleanup",
                         duration_millis_f64(cleanup_elapsed),
                     );
-                    commit_set
-                        .report_old_data_dir_cleanup(&commit_bucket, &commit_object, &old_dir.to_string(), &cleanup)
-                        .await;
                     if (cleanup_ms as u128) >= SET_DISK_COMMIT_TAIL_WARN_THRESHOLD_MS {
                         warn!(
                             event = EVENT_SET_DISK_COMMIT_TAIL_SLOW,
@@ -5681,7 +5761,7 @@ impl SetDisks {
                 if !tail_owns_tmp_cleanup {
                     let cleanup_set = commit_set.clone();
                     let cleanup_tmp_dir = commit_tmp_dir.clone();
-                    tokio::spawn(async move {
+                    cleanup_runtime::spawn(commit_cleanup_reservation.clone(), async move {
                         if let Err(err) = cleanup_set.delete_all(RUSTFS_META_TMP_BUCKET, &cleanup_tmp_dir).await {
                             warn!(tmp_dir = %cleanup_tmp_dir, error = ?err, "failed to cleanup put_object temporary data");
                         } else if issue3031_diag_enabled() {
@@ -5837,7 +5917,7 @@ impl SetDisks {
             // entry is reclaimed by cleanup_stale_tmp_objects (24h expiry,
             // 5-minute background loop).
             let set_disks = self.clone();
-            tokio::spawn(async move {
+            cleanup_runtime::spawn(cleanup_reservation.clone(), async move {
                 if let Err(err) = set_disks.delete_all(RUSTFS_META_TMP_BUCKET, &tmp_dir).await {
                     warn!(tmp_dir = %tmp_dir, error = ?err, "failed to cleanup put_object temporary data");
                 } else if issue3031_diag_enabled() {
@@ -5852,7 +5932,7 @@ impl SetDisks {
             // Failure path (quorum loss / rollback): keep the cleanup inline so
             // a failed PUT never returns while its tmp shards are still on disk
             // (state-residue hardening tracked by backlog#864 / backlog#898).
-            if let Err(err) = self.delete_all(RUSTFS_META_TMP_BUCKET, &tmp_dir).await {
+            if let Err(err) = self.delete_put_tmp_with_budget(&tmp_dir, &cleanup_reservation).await {
                 warn!(tmp_dir = %tmp_dir, error = ?err, "failed to cleanup put_object temporary data");
             } else if issue3031_diag_enabled() {
                 warn!(
@@ -19441,6 +19521,73 @@ mod put_object_tmp_cleanup_tests {
         );
     }
 
+    #[tokio::test]
+    async fn put_cleanup_admission_rejects_before_staging_with_an_external_lock() {
+        const CHILD: &str = "RUSTFS_CLEANUP_ADMISSION_TEST_CHILD";
+        if std::env::var_os(CHILD).is_none() {
+            let output = tokio::task::spawn_blocking(|| {
+                std::process::Command::new(std::env::current_exe().unwrap())
+                    .args([
+                        "--exact",
+                        "set_disk::ops::object::put_object_tmp_cleanup_tests::put_cleanup_admission_rejects_before_staging_with_an_external_lock",
+                        "--nocapture",
+                    ])
+                    .env(CHILD, "1")
+                    .env("RUSTFS_CLEANUP_ISOLATE_ENABLE", "true")
+                    .env("RUSTFS_PUT_RENAME_TAIL_CLEANUP_MAX_PENDING", "1")
+                    .env("RUSTFS_PUT_RENAME_TAIL_CLEANUP_DEFER_WORKERS", "1")
+                    .env_remove("RUSTFS_CLEANUP_CPUS")
+                    .env_remove("RUSTFS_PUT_RENAME_TAIL_CLEANUP_COUNTERFACTUAL_SKIP")
+                    .env_remove("RUSTFS_PUT_RENAME_TAIL_CLEANUP_DEFER_HOLD_WORKER")
+                    .env_remove("RUSTFS_PUT_RENAME_TAIL_CLEANUP_ZERO_TARGET_TMP_DELETE_SKIP")
+                    .output()
+                    .unwrap()
+            })
+            .await
+            .unwrap();
+            assert!(
+                output.status.success(),
+                "{}\n{}",
+                String::from_utf8_lossy(&output.stdout),
+                String::from_utf8_lossy(&output.stderr)
+            );
+            assert!(String::from_utf8_lossy(&output.stdout).contains("1 passed"));
+            return;
+        }
+        let (dirs, disks, set) = hermetic_set_disks(4).await;
+        let bucket = "cleanup-admission-bucket";
+        let object = "externally-locked";
+        for disk in &disks {
+            disk.make_volume(bucket).await.unwrap();
+        }
+        let occupied = cleanup_runtime::reserve().unwrap().expect("isolation enabled in child");
+        let external_lock = set.acquire_write_lock_diag("admission_test", bucket, object).await.unwrap();
+        let mut reader = PutObjReader::from_vec(vec![7u8; TEST_OBJECT_SIZE]);
+        let opts = ObjectOptions {
+            no_lock: true,
+            ..Default::default()
+        };
+        let result = tokio::time::timeout(Duration::from_secs(20), set.put_object(bucket, object, &mut reader, &opts))
+            .await
+            .expect("admission must not wait while the caller owns locks");
+        assert!(matches!(result, Err(StorageError::SlowDown)));
+        assert!(non_trash_tmp_entries(&dirs).await.is_empty(), "rejection must not stage shards");
+        drop(external_lock);
+        drop(occupied);
+        let mut reader = PutObjReader::from_vec(vec![9u8; TEST_OBJECT_SIZE]);
+        set.put_object(bucket, object, &mut reader, &ObjectOptions::default())
+            .await
+            .unwrap();
+        wait_for_tmp_workspace_to_drain(&dirs, "admitted PUT must eventually clean staging").await;
+        let mut actual = set
+            .get_object_reader(bucket, object, None, HeaderMap::new(), &ObjectOptions::default())
+            .await
+            .unwrap();
+        let mut body = Vec::new();
+        actual.stream.read_to_end(&mut body).await.unwrap();
+        assert_eq!(body, vec![9u8; TEST_OBJECT_SIZE]);
+    }
+
     #[tokio::test]
     async fn put_object_success_eventually_cleans_tmp_workspace() {
         let (temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;

crates/ecstore/tests/cleanup_isolation_test.rs

@@ -0,0 +1,131 @@
+// Copyright 2026 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.
+
+#![cfg(target_os = "linux")]
+
+use bytes::Bytes;
+mod storage_api;
+
+use std::time::Duration;
+use storage_api::cleanup_isolation::{DeleteOptions, DiskAPI, DiskError, DiskOption, Endpoint, new_disk};
+
+#[tokio::test]
+async fn cleanup_disk_and_periodic_gc_use_affinity_without_changing_foreground() {
+    const CHILD: &str = "RUSTFS_CLEANUP_LINUX_TEST_CHILD";
+    let foreground = rustix::thread::sched_getaffinity(None).unwrap();
+    let cpu = (0..rustix::thread::CpuSet::MAX_CPU)
+        .find(|cpu| foreground.is_set(*cpu))
+        .expect("the test process must have an allowed CPU");
+    if std::env::var_os(CHILD).is_none() {
+        let output = tokio::task::spawn_blocking(move || {
+            std::process::Command::new(std::env::current_exe().unwrap())
+                .args([
+                    "--exact",
+                    "cleanup_disk_and_periodic_gc_use_affinity_without_changing_foreground",
+                    "--nocapture",
+                ])
+                .env(CHILD, "1")
+                .env("RUSTFS_CLEANUP_ISOLATE_ENABLE", "true")
+                .env("RUSTFS_PUT_RENAME_TAIL_CLEANUP_MAX_PENDING", "4")
+                .env("RUSTFS_PUT_RENAME_TAIL_CLEANUP_DEFER_WORKERS", "1")
+                .env("RUSTFS_CLEANUP_DISK_MAX_PENDING", "4")
+                .env("RUSTFS_CLEANUP_DISK_WORKERS", "1")
+                .env("RUSTFS_CLEANUP_GC_WORKERS", "1")
+                .env("RUSTFS_CLEANUP_BLOCKING_THREADS", "1")
+                .env("RUSTFS_CLEANUP_CPUS", cpu.to_string())
+                .env_remove("RUSTFS_PUT_RENAME_TAIL_CLEANUP_COUNTERFACTUAL_SKIP")
+                .env_remove("RUSTFS_PUT_RENAME_TAIL_CLEANUP_DEFER_HOLD_WORKER")
+                .env_remove("RUSTFS_PUT_RENAME_TAIL_CLEANUP_ZERO_TARGET_TMP_DELETE_SKIP")
+                .output()
+                .unwrap()
+        })
+        .await
+        .unwrap();
+        assert!(
+            output.status.success(),
+            "{}\n{}",
+            String::from_utf8_lossy(&output.stdout),
+            String::from_utf8_lossy(&output.stderr)
+        );
+        assert!(String::from_utf8_lossy(&output.stdout).contains("1 passed"));
+        return;
+    }
+
+    let dir = tempfile::tempdir().unwrap();
+    let disk = new_disk(&Endpoint::try_from(dir.path().to_str().unwrap()).unwrap(), &DiskOption::default())
+        .await
+        .unwrap();
+    let bucket = "cleanup-affinity";
+    disk.make_volume(bucket).await.unwrap();
+    disk.write_all(bucket, "old/part.1", Bytes::from_static(b"garbage"))
+        .await
+        .unwrap();
+    disk.write_all(bucket, "old/.rustfs-old-data-cleanup-receipt.json", Bytes::from_static(b"receipt"))
+        .await
+        .unwrap();
+    disk.write_all(bucket, "live/part.1", Bytes::from_static(b"keep"))
+        .await
+        .unwrap();
+    disk.delete(
+        bucket,
+        "old",
+        DeleteOptions {
+            recursive: true,
+            ..Default::default()
+        },
+    )
+    .await
+    .unwrap();
+    assert!(matches!(disk.read_all(bucket, "old/part.1").await, Err(DiskError::FileNotFound)));
+
+    // One async worker and one lazily started blocking worker. Inspect the
+    // real process threads, including the threads executing filesystem work.
+    let mut cleanup_threads = 0;
+    for entry in std::fs::read_dir("/proc/self/task").unwrap() {
+        let task = entry.unwrap().path();
+        let Ok(name) = std::fs::read_to_string(task.join("comm")) else {
+            continue;
+        };
+        if name.trim() != "rustfs-cleanup" {
+            continue;
+        }
+        cleanup_threads += 1;
+        let status = std::fs::read_to_string(task.join("status")).unwrap();
+        let mask = status
+            .lines()
+            .find_map(|line| line.strip_prefix("Cpus_allowed_list:"))
+            .unwrap()
+            .trim();
+        assert_eq!(mask, cpu.to_string());
+    }
+    assert_eq!(cleanup_threads, 2, "async and blocking workers must both use the configured runtime");
+    assert_eq!(rustix::thread::sched_getaffinity(None).unwrap(), foreground);
+
+    let trash = dir.path().join(".rustfs.sys/tmp/.trash");
+    assert!(std::fs::read_dir(&trash).unwrap().next().is_some());
+    // Advance the foreground timer that dispatches the five-minute periodic
+    // scan; the dedicated runtime then performs the actual physical removal.
+    tokio::time::pause();
+    tokio::time::advance(Duration::from_secs(301)).await;
+    tokio::time::resume();
+    tokio::time::timeout(Duration::from_secs(20), async {
+        while std::fs::read_dir(&trash).unwrap().next().is_some() {
+            tokio::time::sleep(Duration::from_millis(5)).await;
+        }
+    })
+    .await
+    .expect("periodic GC must physically remove trash");
+    assert_eq!(disk.read_all(bucket, "live/part.1").await.unwrap(), Bytes::from_static(b"keep"));
+    assert_eq!(rustix::thread::sched_getaffinity(None).unwrap(), foreground);
+}

crates/ecstore/tests/storage_api.rs

@@ -62,3 +62,8 @@ pub(crate) mod minio_generated_read {
         DiskAPI, DiskOption, Endpoint, Erasure, GetObjectReader, ObjectInfo, ObjectOptions, create_bitrot_reader, new_disk,
     };
 }
+
+pub(crate) mod cleanup_isolation {
+    pub(crate) use super::{DiskAPI, DiskOption, Endpoint, new_disk};
+    pub(crate) use rustfs_ecstore::api::disk::{DeleteOptions, DiskError};
+}

docs/README.md

@@ -8,6 +8,9 @@ collection:
 
 ## Operations
 
+For PUT cleanup admission, blocking-pool isolation, and optional CPU affinity,
+see [PUT cleanup isolation](operations/put-cleanup-isolation.md).
+
 For the logical per-operation io_uring read cap, see
 [io_uring read chunk size](operations/io-uring-read-chunks.md).
 

docs/operations/put-cleanup-isolation.md

@@ -0,0 +1,183 @@
+# PUT and disk cleanup isolation
+
+Cleanup can use process-wide admission budgets and a dedicated Tokio runtime.
+This is opt-in. It preserves cleanup receipts, eligible-data moves to trash,
+PUT temporary workspace cleanup, and periodic physical trash removal. It does
+not disable healing or skip cleanup.
+
+## Configuration
+
+| Variable | Default | Meaning when isolation is enabled |
+| --- | --- | --- |
+| `RUSTFS_CLEANUP_ISOLATE_ENABLE` | `false` | Enable admission and execution isolation on this node. |
+| `RUSTFS_PUT_RENAME_TAIL_CLEANUP_MAX_PENDING` | `1024` | Maximum admitted PUT ownership groups, including surviving commit and cleanup owners. |
+| `RUSTFS_PUT_RENAME_TAIL_CLEANUP_DEFER_WORKERS` | `2` | Maximum active PUT cleanup continuations. Also controls existing deferred queue workers; does not determine OS thread count. |
+| `RUSTFS_CLEANUP_ASYNC_THREADS` | up to `2` | Dedicated async worker threads, limited by available parallelism and an explicit CPU mask by default. Independent of logical job concurrency. |
+| `RUSTFS_CLEANUP_DISK_MAX_PENDING` | `16 × disk workers` (`64`) | Maximum admitted local disk cleanup operations, including queued and running operations. Shared by local requests, incoming RPC cleanup and periodic GC steps. |
+| `RUSTFS_CLEANUP_DISK_WORKERS` | blocking threads (`4`) | Maximum active disk cleanup operations within that budget. |
+| `RUSTFS_CLEANUP_GC_WORKERS` | `1` | Maximum periodic disk scans producing GC steps. |
+| `RUSTFS_CLEANUP_BLOCKING_THREADS` | `4` | Maximum additional blocking threads on the cleanup runtime. |
+| `RUSTFS_CLEANUP_CPUS` | unset | Optional Linux CPU affinity mask, for example `4-5,8`. Applies to async and blocking threads. |
+| `RUSTFS_PUT_RENAME_TAIL_CLEANUP_DEFER_ENABLE` | `false` | Use the existing bounded deferred queue before coordinator execution. Isolation also works without this queue. |
+| `RUSTFS_PUT_RENAME_TAIL_CLEANUP_DEFER_QUEUE_CAPACITY` | `1024` | Capacity of that existing queue; this is not the total admission budget. |
+
+Counts must be positive. PUT workers must not exceed PUT pending capacity;
+GC workers must not exceed disk workers, which must not exceed disk pending
+capacity. Async and blocking thread counts must not exceed 1024. CPU IDs must be supported by the affinity implementation and belong
+to the initiating thread's allowed CPU set. Non-Linux hosts reject explicit CPU
+masks. Configuration is read once at first use, including disk cleanup or GC;
+changes require a restart. Invalid configuration rejects new cleanup admission.
+PUT checks configuration before staging data.
+
+Do not combine isolation with cleanup counterfactuals
+`RUSTFS_PUT_RENAME_TAIL_CLEANUP_COUNTERFACTUAL_SKIP`,
+`RUSTFS_PUT_RENAME_TAIL_CLEANUP_DEFER_HOLD_WORKER`, or
+`RUSTFS_PUT_RENAME_TAIL_CLEANUP_ZERO_TARGET_TMP_DELETE_SKIP`. These combinations
+are rejected because they deliberately suppress cleanup or stop queue progress.
+
+## Ownership and execution
+
+PUT attempts admission before staging. Full admission returns `SlowDown`, so
+clients should retry with backoff. Admission does not wait: COPY readers and
+data-movement callers can already hold locks needed by admitted PUTs. No body
+is ingested or tmp data staged by a rejected PUT.
+
+The request, detached commit owner, and cleanup continuations share one
+reservation. An early ACK or cancelled waiter cannot release it while an owner
+still needs cleanup. After the existing rename-convergence and guard-release
+handoff, deferred workers, queue overflow and direct continuations all acquire
+the same coordinator execution semaphore. Queue overflow cannot bypass that
+limit. Awaited failure cleanup and no-tail tmp cleanup use the same budget.
+Full-tail receipt persistence remains awaited at its existing handoff point;
+old-data reclamation remains after the existing guard release. Incomplete
+rollback still preserves staging for recovery.
+
+Each local disk operation has separate pending and active permits. The wrapper
+below RPC authentication uses this gate for low-level `delete`, `delete_data_dir`
+and old-data cleanup receipt `write_all`. Local PUT fan-out and incoming peer
+RPCs therefore share the same node-local disk budget. Receipt classification
+uses the internal receipt filename; ordinary writes retain their current path.
+RPC schemas and authentication are unchanged. Enable isolation on every node
+to isolate peer execution; settings and CPU masks are not transmitted by RPC.
+
+Coordinator and receiver permits are deliberately distinct: a coordinator can
+wait for peer RPCs without occupying the slots those peers need to service disk
+cleanup. Full disk admission returns an I/O `WouldBlock` error through the
+existing disk error path. A mutation carrying an external publication or
+namespace owner also refuses a full active budget instead of waiting while it
+holds a lock. Existing cleanup error handling and receipt recovery still apply.
+
+Ordinary snapshot-lease release stays on the caller and only updates the in-memory
+registry. It does not allocate a cleanup task or wait for the cleanup scheduler.
+Invalid cleanup configuration also must not strand a reader token: release still
+updates the registry, while any deferred reclamation reports the initialization
+error and keeps both its intent and data. It does not fall back to unbudgeted I/O.
+If the last release triggers deferred old-data
+reclamation, that reclamation tries the same pending and active disk permits.
+Only this case dispatches an owned task to the cleanup runtime.
+Saturation leaves the token released and the cleanup intent/receipt available
+for retry; it does not strand a live reader token or bypass the budget. Quota
+fence release keeps its original receiver lifecycle and does not acquire disk
+execution permits, since it can wait for mutations that already own them.
+
+An admitted request waits for disk execution capacity in the caller's existing
+future. Cancellation or timeout discards this unstarted operation and releases
+pending capacity; it cannot later start deleting a reused path. After execution
+capacity is acquired, the worker also checks whether its result receiver closed
+before its first poll. A job cancelled while waiting for the runtime to schedule
+it is discarded when the worker polls it, releasing its permits without starting
+I/O. Once the worker starts the operation, the owned job retains its permits and
+disk reference until completion. Cancelling or timing out its waiter does not
+free disk capacity while that started job still runs.
+This also retains the disk's mount ownership during detached execution. Cleanup
+waiter timeouts include queue delay and do not by themselves mark a disk faulty;
+the existing active disk-health probe retains its timeout policy.
+
+Periodic trash and stale-tmp scans have a separate scan limit. Each scan acquires
+the shared disk budget to open its directory and then for each entry, releasing
+it between entries. The steps borrow the scan's iterator and paths and run in
+the same owned task, without a spawn/join or path clones per entry. A scan cannot occupy a disk execution slot while waiting
+for another scan slot. GC waits for admission without holding object locks;
+request admission is bounded without creating waiting producer tasks. A single
+trash subtree is still removed as one operation, so a very large subtree can
+hold a slot for its entire removal. Queued scans retain the original mount
+handle even when the corresponding disk object is dropped.
+Missed periodic ticks are skipped so a slow scan does not cause a burst of
+catch-up scans.
+
+Cleanup runs on `rustfs-cleanup` threads. Nested Tokio tasks, Tokio filesystem
+work and ordinary `spawn_blocking` use that runtime. Explicit fsync dispatch
+also stays on its blocking pool. File/directory group-commit keys separate
+cleanup and foreground producers, preventing batches from moving into the
+other class's runtime. Fsync durability settings remain unchanged.
+
+The blocking limit bounds simultaneous blocking execution, not Tokio's internal
+queue or kernel I/O. Disk admission bounds selected operations, not bytes,
+syscall count or device IOPS. Existing synchronous path probes and disk-full
+delete fallbacks can still run on the dedicated async workers, under the disk
+operation budget. All classes still share the filesystem, journal,
+page cache and devices with foreground work.
+
+If CPU binding fails at thread startup, new admissions fail and the failure is
+logged; already-admitted cleanup can finish. Affinity does not reserve exclusive
+cores, move foreground threads elsewhere, change cgroups, or control kernel
+workers and interrupt affinity.
+
+## Scope and recovery
+
+The limits are per process, not cluster-wide quotas. This is not a durable work
+queue. Existing cleanup receipts and tmp/trash recovery retain their role after
+process exit. The option does not bound all storage mutations: metadata/version
+DELETE transactions, heal/scanner coordination and
+multipart orchestration retain their scheduling. Calls from those paths that
+reach the selected disk wrapper entries do share the disk budget; internal
+LocalDisk operations that bypass those entries do not.
+
+## Measurement and tuning
+
+On Linux, `cargo test -p rustfs-ecstore --test cleanup_isolation_test` exercises
+the public disk facade, actual thread affinity, receipt/delete execution and
+periodic physical trash removal with temporary disks. It does not benchmark
+throughput.
+
+With PUT stage metrics enabled, observe:
+
+- `rustfs_cleanup_admitted_puts` and `rustfs_cleanup_admission_rejected_total`;
+- `rustfs_cleanup_execution_wait_seconds`;
+- `rustfs_cleanup_disk_pending`, `rustfs_cleanup_disk_active`, and
+  `rustfs_cleanup_disk_admission_rejected_total`;
+- `rustfs_cleanup_disk_execution_wait_seconds` and `rustfs_cleanup_gc_active`;
+- existing deferred queue depth, age, overflow, completion and service time.
+
+Compare isolation disabled, budget/pool isolation without CPU affinity, and
+isolation with affinity. Keep durability, object sizes, overwrite ratio,
+concurrency and cluster placement fixed. Measure PUT/GET throughput and p95/p99
+latency, rejections, queue age, receipt/trash growth, per-disk I/O latency, CPU
+usage and context switches. Include a long steady-state interval covering
+physical GC, and verify backlog drains afterward.
+
+Start with small worker and blocking-thread limits, then increase only when
+cleanup cannot keep up without harming foreground latency. Too few workers or
+CPUs can increase rejections and lower PUT throughput. Larger queues do not
+increase sustainable device throughput. Affinity helps only when CPU scheduling
+contention matters to the measured workload; no throughput gain is guaranteed.
+
+The defaults protect a node; they are not a claim of optimal production throughput.
+The default disk queue permits sixteen service waves, rather than 256 waves with
+1024 pending operations and four workers. For illustration, at a uniform 100 ms
+per operation, these bounds correspond to roughly 1.6 s versus 25.6 s to drain a
+full queue with no new arrivals. Real service times, stalls and disk fan-out vary;
+there is no queue latency guarantee. The ordinary disk waiter timeout defaults to
+30 s and does not cancel already-admitted work.
+
+Size pending capacity for measured arrival bursts and local/peer fan-out, not
+only for PUT request concurrency. For example, two coordinator jobs each touching
+16 local disks can submit 32 disk operations before incoming peer requests and
+GC are counted. Sixty-four slots need not absorb every distributed burst. Increase
+the explicit limit when measured rejection rates require it, while keeping queue
+age comfortably below waiter deadlines. On many-disk nodes, measure whether four
+active operations underutilize devices before increasing disk/blocking workers.
+The 1024 PUT admission limit includes uploading requests and detached owners;
+it is an object-count bound, not a memory or byte budget. The older deferred queue
+remains optional and disabled by default; its overflow path cannot bypass the
+new coordinator or disk limits. No unbounded foreground fallback is introduced.