From a7064fa2b92948d8fa5a525417fb24908c4817d5 Mon Sep 17 00:00:00 2001 From: GatewayJ <835269233@qq.com> Date: Sun, 27 Sep 2026 16:32:29 +0800 Subject: [PATCH] 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 | 4 +- crates/ecstore/src/disk/cleanup_runtime.rs | 1243 +++++++++++++++++ crates/ecstore/src/disk/disk_store.rs | 379 ++++- crates/ecstore/src/disk/local.rs | 335 +++-- crates/ecstore/src/disk/mod.rs | 1 + crates/ecstore/src/disk/os.rs | 83 +- crates/ecstore/src/set_disk/ops/object.rs | 205 ++- .../ecstore/tests/cleanup_isolation_test.rs | 131 ++ crates/ecstore/tests/storage_api.rs | 5 + docs/README.md | 3 + docs/operations/put-cleanup-isolation.md | 183 +++ 11 files changed, 2437 insertions(+), 135 deletions(-) create mode 100644 crates/ecstore/src/disk/cleanup_runtime.rs create mode 100644 crates/ecstore/tests/cleanup_isolation_test.rs create mode 100644 docs/operations/put-cleanup-isolation.md diff --git a/crates/ecstore/Cargo.toml b/crates/ecstore/Cargo.toml index 4060fc271..de4bad01c 100644 --- a/crates/ecstore/Cargo.toml +++ b/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"] } diff --git a/crates/ecstore/src/disk/cleanup_runtime.rs b/crates/ecstore/src/disk/cleanup_runtime.rs new file mode 100644 index 000000000..7e8744ba9 --- /dev/null +++ b/crates/ecstore/src/disk/cleanup_runtime.rs @@ -0,0 +1,1243 @@ +// 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. + +//! Admission and execution isolation for PUT, local/RPC disk cleanup and GC. +//! PUT admission precedes staging; receiver and GC disk work share a separate +//! budget so coordinators waiting for peers cannot exhaust receiver capacity. + +use crate::disk::{LOG_COMPONENT_ECSTORE, LOG_SUBSYSTEM_DISK}; +use std::collections::BTreeSet; +use std::future::Future; +use std::io; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::{Arc, OnceLock}; +use std::time::Instant; +use tokio::runtime::{Handle, Runtime}; +use tokio::sync::{OwnedSemaphorePermit, Semaphore}; + +const ENABLE: &str = "RUSTFS_CLEANUP_ISOLATE_ENABLE"; +const BUDGET: &str = "RUSTFS_PUT_RENAME_TAIL_CLEANUP_MAX_PENDING"; +const BLOCKING_THREADS: &str = "RUSTFS_CLEANUP_BLOCKING_THREADS"; +const ASYNC_THREADS: &str = "RUSTFS_CLEANUP_ASYNC_THREADS"; +const CPUS: &str = "RUSTFS_CLEANUP_CPUS"; +const WORKERS: &str = "RUSTFS_PUT_RENAME_TAIL_CLEANUP_DEFER_WORKERS"; +const DISK_PENDING: &str = "RUSTFS_CLEANUP_DISK_MAX_PENDING"; +const DISK_WORKERS: &str = "RUSTFS_CLEANUP_DISK_WORKERS"; +const GC_WORKERS: &str = "RUSTFS_CLEANUP_GC_WORKERS"; +const EVENT_CLEANUP_RUNTIME: &str = "disk_cleanup_runtime"; +pub(crate) const OLD_DATA_CLEANUP_RECEIPT_FILE: &str = ".rustfs-old-data-cleanup-receipt.json"; + +#[derive(Clone, Copy)] +enum DiskAdmission { + Request, + Guarded, +} + +#[derive(Clone)] +pub(crate) struct Reservation { + _admission: Arc, + execution: Arc, + handle: Handle, +} + +struct Admission { + _permit: OwnedSemaphorePermit, + metrics_enabled: bool, + metric: &'static str, +} + +impl Admission { + fn new(permit: OwnedSemaphorePermit, metric: &'static str) -> Self { + let metrics_enabled = rustfs_io_metrics::put_stage_metrics_enabled(); + if metrics_enabled { + metrics::gauge!(metric).increment(1.0); + } + Self { + _permit: permit, + metrics_enabled, + metric, + } + } +} + +impl Drop for Admission { + fn drop(&mut self) { + if self.metrics_enabled { + metrics::gauge!(self.metric).decrement(1.0); + } + } +} + +/// A lease release cannot queue behind work that might need the releasing +/// reader's locks. It relinquishes the lease first, then tries this disk budget. +pub(crate) struct DiskExecution { + _active: Admission, + _pending: Admission, +} + +/// Created only inside an owned scan. Steps can borrow the scan's directory +/// iterator and paths: cancelling its waiter cannot cancel a step's syscalls. +#[cfg_attr(test, derive(Default))] +pub(crate) struct GcBudget { + gates: Option<(Arc, Arc)>, +} + +impl GcBudget { + pub(crate) async fn step(&self, future: F) -> super::error::Result + where + F: Future>, + { + let Some((pending, execution)) = &self.gates else { + return future.await; + }; + // Scan producers are bounded and hold no object locks here. Unlike + // requests, they can wait for admission without an unbounded backlog. + let started = rustfs_io_metrics::put_stage_metrics_enabled().then(Instant::now); + let pending = pending.clone().acquire_owned().await.map_err(io::Error::other)?; + execute_disk( + Admission::new(pending, "rustfs_cleanup_disk_pending"), + None, + execution.clone(), + started, + future, + ) + .await + } +} + +fn try_disk_permit(semaphore: &Arc) -> io::Result { + semaphore.clone().try_acquire_owned().map_err(|err| { + if rustfs_io_metrics::put_stage_metrics_enabled() { + metrics::counter!("rustfs_cleanup_disk_admission_rejected_total").increment(1); + } + io::Error::new(io::ErrorKind::WouldBlock, err) + }) +} + +async fn execute_disk( + pending: Admission, + active: Option, + execution: Arc, + started: Option, + future: F, +) -> super::error::Result +where + F: Future>, +{ + let _pending = pending; + let active = match active { + Some(active) => active, + None => execution.acquire_owned().await.map_err(super::error::DiskError::other)?, + }; + let _active = Admission::new(active, "rustfs_cleanup_disk_active"); + if let Some(started) = started { + metrics::histogram!("rustfs_cleanup_disk_execution_wait_seconds").record(started.elapsed().as_secs_f64()); + } + future.await +} + +struct Config { + pending: usize, + workers: usize, + async_threads: usize, + blocking_threads: usize, + cpus: Vec, + disk_pending: usize, + disk_workers: usize, + gc_workers: usize, +} + +fn positive_env(name: &str, default: usize) -> io::Result { + match std::env::var(name) { + Ok(value) => value + .parse::() + .ok() + .filter(|v| *v > 0 && *v <= Semaphore::MAX_PERMITS) + .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidInput, format!("{name} must be a positive supported count"))), + Err(std::env::VarError::NotPresent) => Ok(default), + Err(err) => Err(io::Error::new(io::ErrorKind::InvalidInput, err)), + } +} + +fn parse_cpus(value: &str) -> io::Result> { + let invalid = || { + io::Error::new( + io::ErrorKind::InvalidInput, + "cleanup CPUs must be comma-separated CPU IDs or ascending ranges", + ) + }; + if value.trim().is_empty() { + return Err(invalid()); + } + let mut cpus = BTreeSet::new(); + for part in value.split(',') { + let mut ends = part.trim().split('-'); + let first = ends.next().and_then(|s| s.parse::().ok()).ok_or_else(invalid)?; + let last = match ends.next() { + Some(s) => s.parse::().map_err(|_| invalid())?, + None => first, + }; + if ends.next().is_some() || first > last || last >= 1024 { + return Err(invalid()); + } + cpus.extend(first..=last); + } + Ok(cpus.into_iter().collect()) +} + +impl Config { + fn from_env() -> io::Result { + for name in [ + "RUSTFS_PUT_RENAME_TAIL_CLEANUP_COUNTERFACTUAL_SKIP", + "RUSTFS_PUT_RENAME_TAIL_CLEANUP_DEFER_HOLD_WORKER", + "RUSTFS_PUT_RENAME_TAIL_CLEANUP_ZERO_TARGET_TMP_DELETE_SKIP", + ] { + if rustfs_utils::get_env_bool(name, false) { + return Err(io::Error::new( + io::ErrorKind::InvalidInput, + format!("{name} cannot be combined with cleanup isolation"), + )); + } + } + let workers = positive_env(WORKERS, 2)?; + let pending = positive_env(BUDGET, 1024)?; + let blocking_threads = positive_env(BLOCKING_THREADS, 4)?; + let cpus = match std::env::var(CPUS) { + Ok(value) => parse_cpus(&value)?, + Err(std::env::VarError::NotPresent) => Vec::new(), + Err(err) => return Err(io::Error::new(io::ErrorKind::InvalidInput, err)), + }; + let available = std::thread::available_parallelism().map_or(1, usize::from); + let async_threads = positive_env(ASYNC_THREADS, available.min(if cpus.is_empty() { 2 } else { cpus.len().min(2) }))?; + let disk_workers = positive_env(DISK_WORKERS, blocking_threads)?; + Ok(Self { + pending, + workers, + async_threads, + blocking_threads, + cpus, + // Bound queueing to sixteen service waves by default. Increasing + // async coordinator concurrency does not create more OS threads. + disk_pending: positive_env(DISK_PENDING, disk_workers.saturating_mul(16).min(Semaphore::MAX_PERMITS))?, + disk_workers, + gc_workers: positive_env(GC_WORKERS, 1)?, + }) + } + + fn validate(&self) -> io::Result<()> { + if self.pending == 0 + || self.pending > Semaphore::MAX_PERMITS + || self.workers == 0 + || self.workers > self.pending + || self.async_threads == 0 + || self.async_threads > 1024 + || self.blocking_threads == 0 + || self.blocking_threads > 1024 + || self.disk_pending == 0 + || self.disk_pending > Semaphore::MAX_PERMITS + || self.disk_workers == 0 + || self.disk_workers > self.disk_pending + || self.gc_workers == 0 + || self.gc_workers > self.disk_workers + { + return Err(io::Error::new( + io::ErrorKind::InvalidInput, + "cleanup counts must be positive: PUT workers <= PUT pending, GC workers <= disk workers <= disk pending, async/blocking threads <= 1024", + )); + } + validate_cpus(&self.cpus) + } +} + +#[cfg(target_os = "linux")] +fn validate_cpus(cpus: &[usize]) -> io::Result<()> { + if cpus.is_empty() { + return Ok(()); + } + let allowed = rustix::thread::sched_getaffinity(None)?; + if cpus + .iter() + .any(|cpu| *cpu >= rustix::thread::CpuSet::MAX_CPU || !allowed.is_set(*cpu)) + { + return Err(io::Error::new( + io::ErrorKind::InvalidInput, + "cleanup CPUs must be a subset of the caller's allowed CPU set", + )); + } + Ok(()) +} + +#[cfg(not(target_os = "linux"))] +fn validate_cpus(cpus: &[usize]) -> io::Result<()> { + if cpus.is_empty() { + Ok(()) + } else { + Err(io::Error::new(io::ErrorKind::Unsupported, "cleanup CPU affinity requires Linux")) + } +} + +fn bind_thread(cpus: &[usize]) -> io::Result<()> { + if cpus.is_empty() { + return Ok(()); + } + #[cfg(target_os = "linux")] + { + let mut set = rustix::thread::CpuSet::new(); + for cpu in cpus { + set.set(*cpu); + } + rustix::thread::sched_setaffinity(None, &set)?; + if rustix::thread::sched_getaffinity(None)? != set { + return Err(io::Error::other("cleanup CPU affinity changed during thread startup")); + } + Ok(()) + } + #[cfg(not(target_os = "linux"))] + { + validate_cpus(cpus) + } +} + +struct CleanupRuntime { + runtime: Option, + admission: Arc, + execution: Arc, + affinity_failed: Arc, + disk_admission: Arc, + disk_execution: Arc, + gc_execution: Arc, +} + +impl CleanupRuntime { + fn new(config: Config) -> io::Result { + config.validate()?; + let affinity_failed = Arc::new(AtomicBool::new(false)); + let failed = affinity_failed.clone(); + let cpus = config.cpus; + let runtime = tokio::runtime::Builder::new_multi_thread() + .worker_threads(config.async_threads) + .max_blocking_threads(config.blocking_threads) + .thread_name("rustfs-cleanup") + .enable_all() + .on_thread_start(move || { + // Explicit fsync dispatch must not escape into the shared + // fsync runtime. This callback also runs on blocking threads. + crate::disk::os::use_current_runtime_for_fsync(); + if let Err(err) = bind_thread(&cpus) + && !failed.swap(true, Ordering::AcqRel) + { + tracing::error!(event = EVENT_CLEANUP_RUNTIME, component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_DISK, state = "affinity_failed", error = %err, + "Cleanup thread affinity failed"); + } + }) + .build()?; + Ok(Self { + runtime: Some(runtime), + admission: Arc::new(Semaphore::new(config.pending)), + execution: Arc::new(Semaphore::new(config.workers)), + affinity_failed, + disk_admission: Arc::new(Semaphore::new(config.disk_pending)), + disk_execution: Arc::new(Semaphore::new(config.disk_workers)), + gc_execution: Arc::new(Semaphore::new(config.gc_workers)), + }) + } + + fn reserve(&self) -> io::Result { + if self.affinity_failed.load(Ordering::Acquire) { + return Err(io::Error::other("cleanup thread affinity failed; refusing new cleanup admission")); + } + let metrics_enabled = rustfs_io_metrics::put_stage_metrics_enabled(); + // Callers may already own source-reader or data-movement locks. Do + // not wait here behind a PUT whose completion requires those locks. + let permit = self.admission.clone().try_acquire_owned().map_err(|err| { + if metrics_enabled { + metrics::counter!("rustfs_cleanup_admission_rejected_total").increment(1); + } + io::Error::new(io::ErrorKind::WouldBlock, err) + })?; + let runtime = self + .runtime + .as_ref() + .ok_or_else(|| io::Error::other("cleanup runtime stopped"))?; + Ok(Reservation { + _admission: Arc::new(Admission::new(permit, "rustfs_cleanup_admitted_puts")), + execution: self.execution.clone(), + handle: runtime.handle().clone(), + }) + } + + async fn disk_operation(&self, future: F, mode: DiskAdmission) -> super::error::Result + where + F: Future> + Send + 'static, + T: Send + 'static, + { + if self.affinity_failed.load(Ordering::Acquire) && !super::os::is_isolated_io_thread() { + return Err(io::Error::other("cleanup thread affinity failed; refusing new disk cleanup admission").into()); + } + let started = rustfs_io_metrics::put_stage_metrics_enabled().then(Instant::now); + let pending = try_disk_permit(&self.disk_admission)?; + let pending = Admission::new(pending, "rustfs_cleanup_disk_pending"); + let runtime = self + .runtime + .as_ref() + .ok_or_else(|| io::Error::other("cleanup runtime stopped"))?; + let execution = self.disk_execution.clone(); + // An externally guarded mutation may already hold a namespace lock + // needed by an active disk job. Never queue it behind that job. + let active = if matches!(mode, DiskAdmission::Guarded) { + try_disk_permit(&execution)? + } else { + // Keep queued work in the caller's existing future. A waiter + // timeout must discard an unstarted delete, not leave a task that + // can later delete a reused path after its caller released locks. + execution.clone().acquire_owned().await.map_err(io::Error::other)? + }; + // A permit does not mean the worker has polled this operation. Use + // the result receiver's lifetime to discard work cancelled during + // runtime scheduling, before its first poll can start filesystem I/O. + let (result_tx, result_rx) = tokio::sync::oneshot::channel(); + let task = runtime.spawn(async move { + if result_tx.is_closed() { + return; + } + // After the first poll, the owned job must retain both permits + // through I/O even if the waiter cancels and closes result_rx. + let result = execute_disk(pending, Some(active), execution, started, future).await; + let _ = result_tx.send(result); + }); + match result_rx.await { + Ok(result) => result, + Err(err) => { + // Preserve a panic's JoinError rather than erasing its cause + // into a closed-channel error. + task.await.map_err(super::error::DiskError::other)?; + Err(super::error::DiskError::other(err)) + } + } + } + + fn try_disk_execution(&self) -> io::Result { + let pending = try_disk_permit(&self.disk_admission)?; + let active = try_disk_permit(&self.disk_execution)?; + Ok(DiskExecution { + _pending: Admission::new(pending, "rustfs_cleanup_disk_pending"), + _active: Admission::new(active, "rustfs_cleanup_disk_active"), + }) + } + + async fn gc_operation(&self, scan: F) -> super::error::Result<()> + where + F: FnOnce(GcBudget) -> Fut + Send + 'static, + Fut: Future> + Send + 'static, + { + if self.affinity_failed.load(Ordering::Acquire) { + return Err(io::Error::other("cleanup thread affinity failed; refusing new GC scan").into()); + } + let handle = self + .runtime + .as_ref() + .ok_or_else(|| io::Error::other("cleanup runtime stopped"))? + .handle(); + let execution = self.gc_execution.clone(); + let budget = GcBudget { + gates: Some((self.disk_admission.clone(), self.disk_execution.clone())), + }; + handle + .spawn(async move { + let active = execution.acquire_owned().await.map_err(super::error::DiskError::other)?; + let _active = Admission::new(active, "rustfs_cleanup_gc_active"); + scan(budget).await + }) + .await + .map_err(super::error::DiskError::other)? + } +} + +impl Drop for CleanupRuntime { + fn drop(&mut self) { + if let Some(runtime) = self.runtime.take() { + runtime.shutdown_background(); + } + } +} + +fn instance() -> io::Result> { + static INSTANCE: OnceLock, String>> = OnceLock::new(); + match INSTANCE.get_or_init(|| { + if !rustfs_utils::get_env_bool(ENABLE, false) { + return Ok(None); + } + Config::from_env() + .and_then(CleanupRuntime::new) + .map(Some) + .map_err(|err| err.to_string()) + }) { + Ok(runtime) => Ok(runtime.as_ref()), + Err(err) => Err(io::Error::other(err.clone())), + } +} + +pub(crate) fn reserve() -> io::Result> { + match instance()? { + Some(runtime) => runtime.reserve().map(Some), + None => Ok(None), + } +} + +pub(crate) fn enabled() -> io::Result { + Ok(instance()?.is_some()) +} + +/// Local cleanup entry point, also used by remote RPC receivers. This budget +/// is distinct from coordinator execution: coordinators can wait for peers +/// without exhausting the peers' capacity to service disk cleanup. +pub(crate) async fn disk_operation(future: F) -> super::error::Result +where + F: Future> + Send + 'static, + T: Send + 'static, +{ + match instance()? { + Some(runtime) => runtime.disk_operation(future, DiskAdmission::Request).await, + None => future.await, + } +} + +pub(crate) async fn guarded_disk_operation(future: F) -> super::error::Result +where + F: Future> + Send + 'static, + T: Send + 'static, +{ + match instance()? { + Some(runtime) => runtime.disk_operation(future, DiskAdmission::Guarded).await, + None => future.await, + } +} + +/// Lease release must run even under cleanup saturation, or reader tokens +/// would leak. Only its optional deferred deletion consumes disk permits. +pub(crate) async fn release_lease(future: F) -> super::error::Result<()> +where + F: Future> + Send + 'static, +{ + match instance()? { + Some(runtime) => runtime + .runtime + .as_ref() + .ok_or_else(|| io::Error::other("cleanup runtime stopped"))? + .spawn(future) + .await + .map_err(super::error::DiskError::other)?, + None => future.await, + } +} + +pub(crate) fn try_disk_execution() -> io::Result> { + instance()?.map(CleanupRuntime::try_disk_execution).transpose() +} + +/// At most gc_workers scans can produce work. Each scan yields the shared +/// disk execution budget between entries without spawning a task per entry. +pub(crate) async fn run_gc(scan: F) -> super::error::Result<()> +where + F: FnOnce(GcBudget) -> Fut + Send + 'static, + Fut: Future> + Send + 'static, +{ + match instance()? { + Some(runtime) => runtime.gc_operation(scan).await, + None => scan(GcBudget { gates: None }).await, + } +} + +/// Dispatch only already-admitted work. Keeping the runtime handle in the +/// reservation prevents configuration errors from dropping committed cleanup. +pub(crate) fn spawn(reservation: Option, future: F) -> tokio::task::JoinHandle +where + F: Future + Send + 'static, + F::Output: Send + 'static, +{ + let started = rustfs_io_metrics::put_stage_metrics_enabled().then(Instant::now); + match reservation { + Some(reservation) => reservation.handle.clone().spawn(async move { + // This semaphore is never closed. Both queued work and overflow + // use this gate, and cancellation of the waiter cannot release it. + let _permit = reservation.execution.clone().acquire_owned().await; + if let Some(started) = started { + metrics::histogram!("rustfs_cleanup_execution_wait_seconds").record(started.elapsed().as_secs_f64()); + } + let result = future.await; + drop(reservation); + result + }), + None => tokio::spawn(future), + } +} + +/// Queue workers, queue overflow and synchronous failure cleanup share a gate. +/// With isolation disabled, retain the original inline polling behavior. +pub(crate) async fn run(reservation: Option, future: F) +where + F: Future + Send + 'static, +{ + if reservation.is_none() { + future.await; + } else if let Err(err) = spawn(reservation, future).await { + tracing::error!(event = EVENT_CLEANUP_RUNTIME, component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_DISK, state = "task_failed", error = %err, + "Cleanup task failed"); + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::sync::atomic::AtomicUsize; + use std::time::Duration; + use tokio::sync::oneshot; + + fn config(pending: usize, workers: usize, blocking_threads: usize) -> Config { + Config { + pending, + workers, + async_threads: workers.min(2), + blocking_threads, + cpus: Vec::new(), + disk_pending: 4, + disk_workers: 2, + gc_workers: 1, + } + } + + async fn deadline(future: F) -> F::Output { + tokio::time::timeout(Duration::from_secs(20), future) + .await + .expect("cleanup test timed out") + } + + #[test] + fn cpu_list_rejects_invalid_or_unbounded_ranges() { + assert_eq!(parse_cpus("3, 1-3,6").unwrap(), vec![1, 2, 3, 6]); + for value in ["", " ", "1,", "-1", "2-1", "1-2-3", "no", "1024", "0-18446744073709551615"] { + assert_eq!(parse_cpus(value).unwrap_err().kind(), io::ErrorKind::InvalidInput, "{value}"); + } + } + + #[test] + fn invalid_environment_is_rejected_before_admission() { + const CHILD: &str = "RUSTFS_CLEANUP_CONFIG_TEST_CHILD"; + if std::env::var_os(CHILD).is_some() { + assert!(reserve().is_err()); + return; + } + // Each configuration needs a fresh process: production intentionally + // caches configuration, and mutating process env races parallel tests. + for (name, value) in [ + (WORKERS, "0"), + (WORKERS, "5"), + (BUDGET, "invalid"), + (BLOCKING_THREADS, "0"), + (BLOCKING_THREADS, "1025"), + (ASYNC_THREADS, "0"), + (ASYNC_THREADS, "1025"), + (DISK_PENDING, "0"), + (DISK_WORKERS, "3"), + (GC_WORKERS, "3"), + (CPUS, "1-0"), + ("RUSTFS_PUT_RENAME_TAIL_CLEANUP_COUNTERFACTUAL_SKIP", "true"), + ("RUSTFS_PUT_RENAME_TAIL_CLEANUP_DEFER_HOLD_WORKER", "true"), + ("RUSTFS_PUT_RENAME_TAIL_CLEANUP_ZERO_TARGET_TMP_DELETE_SKIP", "true"), + ] { + let result = std::process::Command::new(std::env::current_exe().unwrap()) + .args([ + "--exact", + "disk::cleanup_runtime::tests::invalid_environment_is_rejected_before_admission", + "--nocapture", + ]) + .env(CHILD, "1") + .env(ENABLE, "true") + .env(WORKERS, "1") + .env(BUDGET, "4") + .env(BLOCKING_THREADS, "1") + .env(ASYNC_THREADS, "1") + .env(DISK_PENDING, "2") + .env(DISK_WORKERS, "2") + .env(GC_WORKERS, "1") + .env_remove(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") + .env(name, value) + .output() + .unwrap(); + assert!(result.status.success(), "{name}={value}: {}", String::from_utf8_lossy(&result.stderr)); + assert!( + String::from_utf8_lossy(&result.stdout).contains("1 passed"), + "child must execute the admission assertion" + ); + } + } + + #[test] + fn invalid_budgets_fail_before_building_threads() { + for config in [config(0, 1, 1), config(1, 0, 1), config(1, 2, 1), config(1, 1, 0)] { + assert!(matches!(CleanupRuntime::new(config), Err(err) if err.kind() == io::ErrorKind::InvalidInput)); + } + } + + #[tokio::test] + async fn disk_work_keeps_budget_after_rpc_waiter_cancellation() { + let mut settings = config(1, 1, 1); + settings.disk_pending = 1; + settings.disk_workers = 1; + let runtime = Arc::new(CleanupRuntime::new(settings).unwrap()); + let receiver = runtime.clone(); + let (entered, entry) = oneshot::channel(); + let (release, released) = std::sync::mpsc::channel(); + let rpc = tokio::spawn(async move { + receiver + .disk_operation( + async move { + tokio::task::spawn_blocking(move || { + assert_eq!(std::thread::current().name(), Some("rustfs-cleanup")); + entered.send(()).unwrap(); + released.recv_timeout(Duration::from_secs(20)).unwrap(); + }) + .await + .unwrap(); + Ok(()) + }, + DiskAdmission::Request, + ) + .await + }); + deadline(entry).await.unwrap(); + rpc.abort(); + assert!(rpc.await.unwrap_err().is_cancelled()); + assert_eq!(runtime.disk_admission.available_permits(), 0); + assert_eq!(runtime.disk_execution.available_permits(), 0); + let ran = Arc::new(AtomicBool::new(false)); + let invoked = ran.clone(); + let rejected = runtime + .disk_operation( + async move { + invoked.store(true, Ordering::SeqCst); + Ok(()) + }, + DiskAdmission::Request, + ) + .await; + assert!(matches!(rejected, Err(super::super::error::DiskError::Io(err)) if err.kind() == io::ErrorKind::WouldBlock)); + assert!(!ran.load(Ordering::SeqCst), "rejected work must not execute"); + release.send(()).unwrap(); + deadline(async { + while runtime.disk_admission.available_permits() == 0 { + tokio::task::yield_now().await; + } + }) + .await; + assert_eq!(runtime.disk_execution.available_permits(), 1); + } + + #[tokio::test] + async fn cancelling_queued_disk_work_discards_it_before_execution() { + let runtime = Arc::new(CleanupRuntime::new(config(1, 1, 1)).expect("cleanup runtime")); + let occupied = runtime + .disk_execution + .clone() + .acquire_many_owned(2) + .await + .expect("occupy all execution slots"); + let receiver = runtime.clone(); + let ran = Arc::new(AtomicBool::new(false)); + let invoked = ran.clone(); + let waiter = tokio::spawn(async move { + receiver + .disk_operation( + async move { + invoked.store(true, Ordering::SeqCst); + Ok(()) + }, + DiskAdmission::Request, + ) + .await + }); + deadline(async { + while runtime.disk_admission.available_permits() == 4 { + tokio::task::yield_now().await; + } + }) + .await; + waiter.abort(); + assert!(waiter.await.expect_err("cancel queued waiter").is_cancelled()); + assert_eq!( + runtime.disk_admission.available_permits(), + 4, + "unstarted work must release pending capacity immediately" + ); + drop(occupied); + deadline(runtime.disk_operation(async { Ok(()) }, DiskAdmission::Request)) + .await + .expect("later work must progress"); + assert!(!ran.load(Ordering::SeqCst), "cancelled queued deletion must never execute later"); + } + + #[tokio::test] + async fn cancelling_disk_work_before_worker_poll_preserves_reused_path() { + let runtime = Arc::new(CleanupRuntime::new(config(1, 1, 1)).expect("cleanup runtime")); + let (entered, entry) = oneshot::channel(); + let (release, released) = std::sync::mpsc::channel(); + let stalled = runtime.runtime.as_ref().expect("live runtime").spawn(async move { + entered.send(()).expect("worker entered"); + released + .recv_timeout(Duration::from_secs(30)) + .expect("release stalled worker"); + }); + deadline(entry).await.expect("worker must stall before dispatch"); + let dir = tempfile::tempdir().expect("temporary disk"); + let backup = dir.path().join("xl.meta.bkp"); + std::fs::write(&backup, b"previous transaction").expect("old backup"); + { + let path = backup.clone(); + let deletion = runtime.disk_operation( + async move { + tokio::fs::remove_file(path).await?; + Ok(()) + }, + DiskAdmission::Request, + ); + tokio::pin!(deletion); + assert!(futures::poll!(&mut deletion).is_pending()); + assert_eq!(runtime.disk_execution.available_permits(), 1, "execution was admitted but not polled"); + } + std::fs::write(&backup, b"next transaction").expect("reuse path after cancelled waiter"); + release.send(()).expect("resume worker"); + deadline(stalled).await.expect("worker resumed"); + deadline(async { + while runtime.disk_admission.available_permits() != 4 { + tokio::task::yield_now().await; + } + }) + .await; + assert_eq!( + std::fs::read(backup).expect("unstarted deletion must not execute after cancellation"), + b"next transaction" + ); + assert_eq!(runtime.disk_execution.available_permits(), 2); + } + + #[tokio::test] + async fn disk_task_panic_preserves_cause_and_releases_budget() { + async fn panicking_job() -> super::super::error::Result<()> { + panic!("injected cleanup panic"); + } + let runtime = CleanupRuntime::new(config(1, 1, 1)).expect("cleanup runtime"); + let error = deadline(runtime.disk_operation(panicking_job(), DiskAdmission::Request)) + .await + .expect_err("worker panic must reach its caller"); + let super::super::error::DiskError::Io(error) = error else { + panic!("worker panic must retain its I/O error source"); + }; + let cause = error + .get_ref() + .expect("panic source") + .downcast_ref::() + .expect("preserved JoinError"); + assert!(cause.is_panic()); + assert_eq!(runtime.disk_admission.available_permits(), 4); + assert_eq!(runtime.disk_execution.available_permits(), 2); + } + + #[tokio::test] + async fn coordinator_waiting_for_disk_does_not_consume_receiver_capacity() { + let mut settings = config(1, 1, 1); + settings.disk_pending = 1; + settings.disk_workers = 1; + let runtime = Arc::new(CleanupRuntime::new(settings).unwrap()); + let reservation = runtime.reserve().unwrap(); + let receiver = runtime.clone(); + let result = deadline(spawn(Some(reservation), async move { + receiver + .disk_operation(async { Ok(42) }, DiskAdmission::Request) + .await + .unwrap() + })) + .await + .unwrap(); + assert_eq!(result, 42); + } + + #[tokio::test] + async fn gc_shares_disk_budget_and_guarded_requests_do_not_wait_under_locks() { + let mut settings = config(1, 1, 1); + settings.disk_pending = 1; + settings.disk_workers = 1; + let runtime = Arc::new(CleanupRuntime::new(settings).unwrap()); + let receiver = runtime.clone(); + let (entered, entry) = oneshot::channel(); + let (release, released) = oneshot::channel(); + let rpc = tokio::spawn(async move { + receiver + .disk_operation( + async move { + entered.send(()).unwrap(); + released.await.unwrap(); + Ok(()) + }, + DiskAdmission::Request, + ) + .await + }); + deadline(entry).await.unwrap(); + let guarded = runtime.disk_operation(async { Ok(()) }, DiskAdmission::Guarded).await; + assert!(matches!(guarded, Err(super::super::error::DiskError::Io(err)) if err.kind() == io::ErrorKind::WouldBlock)); + let budget = GcBudget { + gates: Some((runtime.disk_admission.clone(), runtime.disk_execution.clone())), + }; + let gc = budget.step(async { Ok(7) }); + tokio::pin!(gc); + assert!( + futures::poll!(&mut gc).is_pending(), + "GC must join the shared budget rather than bypass it" + ); + release.send(()).unwrap(); + deadline(rpc).await.unwrap().unwrap(); + assert_eq!(deadline(gc).await.unwrap(), 7); + + // Also reject a guarded mutation when pending capacity is free but + // execution is occupied. This prevents namespace -> budget inversion. + let execution = runtime.disk_execution.clone().acquire_owned().await.unwrap(); + let guarded = runtime.disk_operation(async { Ok(()) }, DiskAdmission::Guarded).await; + assert!(matches!(guarded, Err(super::super::error::DiskError::Io(err)) if err.kind() == io::ErrorKind::WouldBlock)); + assert_eq!(runtime.disk_admission.available_permits(), 1); + assert!(matches!(runtime.try_disk_execution(), Err(err) if err.kind() == io::ErrorKind::WouldBlock)); + assert_eq!(runtime.disk_admission.available_permits(), 1); + drop(execution); + let lease_cleanup = runtime.try_disk_execution().unwrap(); + assert_eq!(runtime.disk_admission.available_permits(), 0); + assert_eq!(runtime.disk_execution.available_permits(), 0); + drop(lease_cleanup); + assert_eq!(runtime.disk_execution.available_permits(), 1); + } + + #[tokio::test] + async fn gc_scan_limit_leaves_disk_slots_available_between_entries() { + let runtime = Arc::new(CleanupRuntime::new(config(1, 1, 1)).unwrap()); + let first_runtime = runtime.clone(); + let (entered, entry) = oneshot::channel(); + let (release, released) = oneshot::channel(); + let first = tokio::spawn(async move { + first_runtime + .gc_operation(move |_| async move { + entered.send(()).unwrap(); + released.await.unwrap(); + Ok(()) + }) + .await + }); + deadline(entry).await.unwrap(); + assert_eq!(runtime.gc_execution.available_permits(), 0); + assert_eq!( + deadline(runtime.disk_operation(async { Ok(42) }, DiskAdmission::Request)) + .await + .unwrap(), + 42 + ); + let second = runtime.gc_operation(|_| async { Ok(()) }); + tokio::pin!(second); + assert!(futures::poll!(&mut second).is_pending()); + release.send(()).unwrap(); + deadline(first).await.unwrap().unwrap(); + deadline(second).await.unwrap(); + } + + #[tokio::test] + async fn gc_steps_borrow_scan_state_and_keep_budget_after_waiter_cancellation() { + let runtime = Arc::new(CleanupRuntime::new(config(1, 1, 1)).expect("cleanup runtime")); + let receiver = runtime.clone(); + let (entered, entry) = oneshot::channel(); + let (release, released) = std::sync::mpsc::channel(); + let waiter = tokio::spawn(async move { + receiver + .gc_operation(move |budget| async move { + let scan_task = tokio::task::id(); + let mut count = 0; + for _ in 0..32 { + budget + .step(async { + assert_eq!(tokio::task::id(), scan_task, "each entry must stay on the owned scan task"); + count += 1; + Ok(()) + }) + .await?; + } + assert_eq!(count, 32); + budget + .step(async move { + tokio::task::spawn_blocking(move || { + entered.send(()).expect("notify started syscall"); + released.recv_timeout(Duration::from_secs(20)).expect("release syscall"); + }) + .await + .expect("blocking cleanup"); + Ok(()) + }) + .await + }) + .await + }); + deadline(entry).await.expect("GC must start"); + waiter.abort(); + assert!(waiter.await.expect_err("cancelled waiter").is_cancelled()); + assert_eq!(runtime.gc_execution.available_permits(), 0); + assert_eq!(runtime.disk_admission.available_permits(), 3); + assert_eq!(runtime.disk_execution.available_permits(), 1); + release.send(()).expect("release blocking cleanup"); + deadline(async { + while runtime.gc_execution.available_permits() == 0 { + tokio::task::yield_now().await; + } + }) + .await; + assert_eq!(runtime.disk_admission.available_permits(), 4); + assert_eq!(runtime.disk_execution.available_permits(), 2); + } + + #[test] + fn defaults_bound_queue_waves_and_separate_threads_from_jobs() { + const CHILD: &str = "RUSTFS_CLEANUP_DEFAULTS_TEST_CHILD"; + if std::env::var_os(CHILD).is_some() { + let config = Config::from_env().expect("default configuration"); + assert_eq!(config.workers, 64); + assert!(config.async_threads <= 2); + assert_eq!(config.blocking_threads, 4); + assert_eq!(config.disk_workers, 4); + assert_eq!(config.disk_pending, 64); + config.validate().expect("defaults must be internally consistent"); + let threads = config.async_threads; + let runtime = CleanupRuntime::new(config).expect("runtime with many logical jobs"); + assert_eq!(runtime.runtime.as_ref().expect("live runtime").metrics().num_workers(), threads); + return; + } + let output = std::process::Command::new(std::env::current_exe().expect("test executable")) + .args([ + "--exact", + "disk::cleanup_runtime::tests::defaults_bound_queue_waves_and_separate_threads_from_jobs", + ]) + .env(CHILD, "1") + .env(WORKERS, "64") + .env(BUDGET, "1024") + .env(GC_WORKERS, "1") + .env_remove(ASYNC_THREADS) + .env_remove(BLOCKING_THREADS) + .env_remove(DISK_WORKERS) + .env_remove(DISK_PENDING) + .env_remove(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() + .expect("run defaults test in a fresh process"); + assert!(output.status.success(), "{}", String::from_utf8_lossy(&output.stderr)); + assert!(String::from_utf8_lossy(&output.stdout).contains("1 passed")); + } + + #[tokio::test] + async fn affinity_failure_rejects_new_admission_without_dropping_admitted_work() { + let runtime = CleanupRuntime::new(config(2, 1, 1)).unwrap(); + let reservation = runtime.reserve().unwrap(); + runtime.affinity_failed.store(true, Ordering::Release); + assert!(runtime.reserve().is_err()); + assert_eq!(deadline(spawn(Some(reservation), async { 42 })).await.unwrap(), 42); + assert_eq!(runtime.admission.available_permits(), 2); + } + + #[tokio::test] + async fn admission_is_retained_by_every_commit_and_cleanup_owner() { + let runtime = CleanupRuntime::new(config(1, 1, 1)).unwrap(); + let request = runtime.reserve().unwrap(); + let commit = request.clone(); + let (release, released) = oneshot::channel(); + let cleanup = spawn(Some(commit.clone()), async move { + released.await.unwrap(); + }); + drop(request); + drop(commit); + assert!( + matches!(runtime.reserve(), Err(err) if err.kind() == io::ErrorKind::WouldBlock), + "queued and running cleanup must retain admission without waiting on external locks" + ); + release.send(()).unwrap(); + deadline(cleanup).await.unwrap(); + let next = runtime.reserve().unwrap(); + assert_eq!(runtime.admission.available_permits(), 0); + drop(next); + assert_eq!(runtime.admission.available_permits(), 1); + } + + #[tokio::test] + async fn cancelling_cleanup_waiter_does_not_cancel_work_or_release_budget() { + let runtime = CleanupRuntime::new(config(1, 1, 1)).unwrap(); + let reservation = runtime.reserve().unwrap(); + let (entered, entry) = oneshot::channel(); + let (release, released) = oneshot::channel(); + let waiter = tokio::spawn(run(Some(reservation), async move { + entered.send(()).unwrap(); + released.await.unwrap(); + })); + deadline(entry).await.unwrap(); + waiter.abort(); + assert!(waiter.await.unwrap_err().is_cancelled()); + assert_eq!(runtime.admission.available_permits(), 0); + assert_eq!(runtime.execution.available_permits(), 0); + release.send(()).unwrap(); + deadline(async { + loop { + if let Ok(reservation) = runtime.reserve() { + drop(reservation); + break; + } + tokio::task::yield_now().await; + } + }) + .await; + assert_eq!(runtime.admission.available_permits(), 1); + } + + #[tokio::test] + async fn worker_and_overflow_dispatch_share_one_execution_budget() { + let runtime = CleanupRuntime::new(config(8, 2, 1)).unwrap(); + let active = Arc::new(AtomicUsize::new(0)); + let peak = Arc::new(AtomicUsize::new(0)); + let gate = Arc::new(Semaphore::new(0)); + let (entered, mut entries) = tokio::sync::mpsc::unbounded_channel(); + let mut jobs = Vec::new(); + for index in 0..8 { + let reservation = runtime.reserve().unwrap(); + let active = active.clone(); + let peak = peak.clone(); + let gate = gate.clone(); + let entered = entered.clone(); + let work = async move { + let running = active.fetch_add(1, Ordering::SeqCst) + 1; + peak.fetch_max(running, Ordering::SeqCst); + entered.send(()).unwrap(); + gate.acquire().await.unwrap().forget(); + active.fetch_sub(1, Ordering::SeqCst); + }; + // Both a dequeued job and a queue-full continuation enter `run`; + // no-tail temporary cleanup uses direct owned dispatch. + jobs.push(if index % 2 == 0 { + tokio::spawn(run(Some(reservation), work)) + } else { + spawn(Some(reservation), work) + }); + } + deadline(entries.recv()).await.unwrap(); + deadline(entries.recv()).await.unwrap(); + assert_eq!(runtime.execution.available_permits(), 0); + assert!(entries.try_recv().is_err()); + assert_eq!(runtime.admission.available_permits(), 0); + gate.add_permits(8); + for job in jobs { + deadline(job).await.unwrap(); + } + assert_eq!(peak.load(Ordering::SeqCst), 2); + assert_eq!(runtime.admission.available_permits(), 8); + } + + #[tokio::test] + async fn cleanup_children_and_blocking_work_use_dedicated_threads() { + let runtime = CleanupRuntime::new(config(1, 1, 1)).unwrap(); + let task = spawn(Some(runtime.reserve().unwrap()), async move { + assert_eq!(std::thread::current().name(), Some("rustfs-cleanup")); + tokio::spawn(async { + assert_eq!(std::thread::current().name(), Some("rustfs-cleanup")); + tokio::task::spawn_blocking(|| { + assert_eq!(std::thread::current().name(), Some("rustfs-cleanup")); + }) + .await + .unwrap(); + }) + .await + .unwrap(); + }); + deadline(task).await.unwrap(); + assert_ne!(std::thread::current().name(), Some("rustfs-cleanup")); + } + + #[tokio::test] + async fn blocking_pool_saturation_does_not_block_foreground_pool() { + let runtime = CleanupRuntime::new(config(2, 2, 1)).unwrap(); + let (entered, entry) = oneshot::channel(); + let (release, released) = std::sync::mpsc::channel(); + let first = spawn(Some(runtime.reserve().unwrap()), async move { + tokio::task::spawn_blocking(move || { + entered.send(()).unwrap(); + released.recv_timeout(Duration::from_secs(20)).unwrap(); + }) + .await + .unwrap(); + }); + deadline(entry).await.unwrap(); + let (submitted, submission) = oneshot::channel(); + let (second_entered, mut second_entry) = oneshot::channel(); + let second = spawn(Some(runtime.reserve().unwrap()), async move { + let task = tokio::task::spawn_blocking(move || { + second_entered.send(()).unwrap(); + }); + submitted.send(()).unwrap(); + task.await.unwrap(); + }); + deadline(submission).await.unwrap(); + assert!(matches!(second_entry.try_recv(), Err(oneshot::error::TryRecvError::Empty))); + assert_eq!(deadline(tokio::task::spawn_blocking(|| 42)).await.unwrap(), 42); + release.send(()).unwrap(); + deadline(first).await.unwrap(); + deadline(second).await.unwrap(); + deadline(second_entry).await.unwrap(); + } + + #[cfg(target_os = "linux")] + #[tokio::test] + async fn affinity_covers_async_and_blocking_threads_without_changing_caller() { + let caller = rustix::thread::sched_getaffinity(None).unwrap(); + let cpu = (0..rustix::thread::CpuSet::MAX_CPU).find(|cpu| caller.is_set(*cpu)).unwrap(); + let mut config = config(1, 1, 1); + config.cpus = vec![cpu]; + let runtime = CleanupRuntime::new(config).unwrap(); + deadline(spawn(Some(runtime.reserve().unwrap()), async move { + let expected = { + let mut set = rustix::thread::CpuSet::new(); + set.set(cpu); + set + }; + assert_eq!(rustix::thread::sched_getaffinity(None).unwrap(), expected); + let actual = tokio::task::spawn_blocking(|| rustix::thread::sched_getaffinity(None).unwrap()) + .await + .unwrap(); + assert_eq!(actual, expected); + })) + .await + .unwrap(); + assert_eq!(rustix::thread::sched_getaffinity(None).unwrap(), caller); + assert!(validate_cpus(&[rustix::thread::CpuSet::MAX_CPU]).is_err()); + } + + #[cfg(not(target_os = "linux"))] + #[test] + fn explicit_cpu_affinity_is_rejected_on_unsupported_platforms() { + assert_eq!(validate_cpus(&[0]).unwrap_err().kind(), io::ErrorKind::Unsupported); + validate_cpus(&[]).unwrap(); + } +} diff --git a/crates/ecstore/src/disk/disk_store.rs b/crates/ecstore/src/disk/disk_store.rs index 8c4fb2a72..6e521cec9 100644 --- a/crates/ecstore/src/disk/disk_store.rs +++ b/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>, ) -> 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>, + ) -> 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( + &self, + op: &'static str, + mutation: DiskMetricMutation, + operation: F, + timeout_duration: Duration, + ) -> Result + where + F: FnOnce() -> Fut, + Fut: std::future::Future>, + { + // 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( &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::>(), + 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)); diff --git a/crates/ecstore/src/disk/local.rs b/crates/ecstore/src/disk/local.rs index 8888d4102..89e03caaa 100644 --- a/crates/ecstore/src/disk/local.rs +++ b/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, #[cfg(target_os = "linux")] mount_lease_mount_id: Option, /// 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, 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(®ular_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" diff --git a/crates/ecstore/src/disk/mod.rs b/crates/ecstore/src/disk/mod.rs index e4e2d8f10..6d219c5ee 100644 --- a/crates/ecstore/src/disk/mod.rs +++ b/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; diff --git a/crates/ecstore/src/disk/os.rs b/crates/ecstore/src/disk/os.rs index 14a56b598..e8127b6a2 100644 --- a/crates/ecstore/src/disk/os.rs +++ b/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, @@ -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> = LazyLock::new( } }); +thread_local! { + static FSYNC_ON_CURRENT_RUNTIME: std::cell::Cell = 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(f: impl FnOnce() -> T + Send + 'static) -> tokio::task::JoinHandle { + 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 { Arc::new(Semaphore::new(MAX_PARALLEL_FILE_SYNCS)) } diff --git a/crates/ecstore/src/set_disk/ops/object.rs b/crates/ecstore/src/set_disk/ops/object.rs index aad2bf42b..3867fcfb3 100644 --- a/crates/ecstore/src/set_disk/ops/object.rs +++ b/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) -> 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, @@ -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; diff --git a/crates/ecstore/tests/cleanup_isolation_test.rs b/crates/ecstore/tests/cleanup_isolation_test.rs new file mode 100644 index 000000000..840cd7151 --- /dev/null +++ b/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); +} diff --git a/crates/ecstore/tests/storage_api.rs b/crates/ecstore/tests/storage_api.rs index 1c2ef784b..84717f088 100644 --- a/crates/ecstore/tests/storage_api.rs +++ b/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}; +} diff --git a/docs/README.md b/docs/README.md index 3a91c4278..30c870b4e 100644 --- a/docs/README.md +++ b/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). diff --git a/docs/operations/put-cleanup-isolation.md b/docs/operations/put-cleanup-isolation.md new file mode 100644 index 000000000..ad874db39 --- /dev/null +++ b/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.