From 1de0ba916d50f762f5731b870db308f07175d584 Mon Sep 17 00:00:00 2001 From: houseme Date: Thu, 28 May 2026 23:14:30 +0800 Subject: [PATCH] fix(ecstore): reduce restart-time startup race noise (#3108) Treat empty ping bodies as liveness probes, add a startup cleanup barrier for early walk_dir calls, and delay immediate background cleanup/capacity timers to reduce transient restart-time VolumeNotFound noise. Also downgrade expected missing-path producer results during startup from generic errors to warnings while preserving existing storage semantics. --- .../ecstore/src/cache_value/metacache_set.rs | 6 +- crates/ecstore/src/disk/local.rs | 134 +++++++++++++++++- .../object-capacity/src/capacity_manager.rs | 4 +- rustfs/src/storage/rpc/node_service.rs | 29 +++- 4 files changed, 159 insertions(+), 14 deletions(-) diff --git a/crates/ecstore/src/cache_value/metacache_set.rs b/crates/ecstore/src/cache_value/metacache_set.rs index 10561443f..0718c470a 100644 --- a/crates/ecstore/src/cache_value/metacache_set.rs +++ b/crates/ecstore/src/cache_value/metacache_set.rs @@ -492,7 +492,11 @@ pub async fn list_path_raw(rx: CancellationToken, opts: ListPathRawOptions) -> d match result { Ok(Ok(())) => {} Ok(Err(err)) => { - error!("list_path_raw producer err {:?}", err); + if matches!(err, DiskError::FileNotFound | DiskError::VolumeNotFound) { + warn!("list_path_raw producer missing path {:?}", err); + } else { + error!("list_path_raw producer err {:?}", err); + } job_errs.push(err); } Err(err) => { diff --git a/crates/ecstore/src/disk/local.rs b/crates/ecstore/src/disk/local.rs index a4357f998..0890cfdc5 100644 --- a/crates/ecstore/src/disk/local.rs +++ b/crates/ecstore/src/disk/local.rs @@ -57,14 +57,15 @@ use std::{ use time::OffsetDateTime; use tokio::fs::{self, File}; use tokio::io::{AsyncReadExt, AsyncSeekExt, AsyncWrite, AsyncWriteExt, ErrorKind}; -use tokio::sync::RwLock; -use tokio::time::interval; +use tokio::sync::{Notify, RwLock}; +use tokio::time::{Instant, interval_at, timeout}; use tracing::{debug, error, info, warn}; use uuid::Uuid; const DELETED_OBJECTS_CLEANUP_INTERVAL: Duration = Duration::from_secs(60 * 5); const STALE_TMP_OBJECT_EXPIRY: Duration = Duration::from_secs(24 * 60 * 60); const RUSTFS_META_TMP_OLD_BUCKET: &str = ".rustfs.sys/tmp-old"; +const STARTUP_CLEANUP_WAIT_TIMEOUT: Duration = Duration::from_secs(2); #[derive(Debug, Clone)] pub struct FormatInfo { @@ -331,6 +332,8 @@ pub struct LocalDisk { // pub format_data: Mutex>, // pub format_file_info: Mutex>, // pub format_last_check: Mutex>, + startup_cleanup_ready: Arc, + startup_cleanup_notify: Arc, exit_signal: Option>, } @@ -370,7 +373,15 @@ impl LocalDisk { ensure_data_usage_layout(&root).await.map_err(DiskError::from)?; - if cleanup && let Err(err) = Self::cleanup_tmp_on_startup(&root).await { + let startup_cleanup_ready = Arc::new(AtomicU32::new(u32::from(!cleanup))); + let startup_cleanup_notify = Arc::new(Notify::new()); + + if cleanup + && let Err(err) = + Self::cleanup_tmp_on_startup(&root, startup_cleanup_ready.clone(), startup_cleanup_notify.clone()).await + { + startup_cleanup_ready.store(1, Ordering::Release); + startup_cleanup_notify.notify_waiters(); warn!(root = ?root, error = ?err, "failed to cleanup temporary data during disk startup"); } @@ -466,6 +477,8 @@ impl LocalDisk { // format_last_check: Mutex::new(format_last_check), path_cache: Arc::new(ParkingLotRwLock::new(HashMap::with_capacity(2048))), current_dir: Arc::new(OnceLock::new()), + startup_cleanup_ready, + startup_cleanup_notify, exit_signal: None, }; let (info, _root) = get_disk_info(root).await?; @@ -497,7 +510,8 @@ impl LocalDisk { } async fn cleanup_deleted_objects_loop(root: PathBuf, mut exit_rx: tokio::sync::broadcast::Receiver<()>) { - let mut interval = interval(DELETED_OBJECTS_CLEANUP_INTERVAL); + let start_at = Instant::now() + DELETED_OBJECTS_CLEANUP_INTERVAL; + let mut interval = interval_at(start_at, DELETED_OBJECTS_CLEANUP_INTERVAL); loop { tokio::select! { _ = interval.tick() => { @@ -525,12 +539,18 @@ impl LocalDisk { root.join(meta_path) } - async fn cleanup_tmp_on_startup(root: &Path) -> Result<()> { + async fn cleanup_tmp_on_startup( + root: &Path, + startup_cleanup_ready: Arc, + startup_cleanup_notify: Arc, + ) -> Result<()> { let tmp_path = Self::meta_path(root, RUSTFS_META_TMP_BUCKET); let tmp_old_path = Self::meta_path(root, RUSTFS_META_TMP_OLD_BUCKET).join(Uuid::new_v4().to_string()); rename_all(&tmp_path, &tmp_old_path, root).await?; + tokio::fs::create_dir_all(Self::meta_path(root, RUSTFS_META_TMP_DELETED_BUCKET)).await?; + let tmp_old_root = Self::meta_path(root, RUSTFS_META_TMP_OLD_BUCKET); tokio::spawn(async move { if let Err(err) = tokio::fs::remove_dir_all(&tmp_old_root).await @@ -538,12 +558,35 @@ impl LocalDisk { { warn!(path = ?tmp_old_root, error = ?err, "failed to remove old temporary data"); } + startup_cleanup_ready.store(1, Ordering::Release); + startup_cleanup_notify.notify_waiters(); }); - tokio::fs::create_dir_all(Self::meta_path(root, RUSTFS_META_TMP_DELETED_BUCKET)).await?; Ok(()) } + async fn wait_for_startup_cleanup(&self) { + if self.startup_cleanup_ready.load(Ordering::Acquire) != 0 { + return; + } + + if wait_for_startup_cleanup_signal( + self.startup_cleanup_ready.as_ref(), + self.startup_cleanup_notify.as_ref(), + STARTUP_CLEANUP_WAIT_TIMEOUT, + ) + .await + { + debug!(disk = %self.endpoint, "startup cleanup barrier released before walk_dir"); + } else { + warn!( + disk = %self.endpoint, + timeout_ms = STARTUP_CLEANUP_WAIT_TIMEOUT.as_millis(), + "startup cleanup barrier timed out; continuing walk_dir" + ); + } + } + async fn cleanup_stale_tmp_objects(root: PathBuf) -> Result<()> { Self::cleanup_stale_tmp_objects_with_expiry(root, STALE_TMP_OBJECT_EXPIRY).await } @@ -2375,6 +2418,8 @@ impl DiskAPI for LocalDisk { // FIXME: TODO: io.writer TODO cancel #[tracing::instrument(level = "debug", skip(self, wr))] async fn walk_dir(&self, opts: WalkDirOptions, wr: &mut W) -> Result<()> { + self.wait_for_startup_cleanup().await; + let volume_dir = self.get_bucket_path(&opts.bucket)?; if !skip_access_checks(&opts.bucket) @@ -3105,6 +3150,31 @@ impl DiskAPI for LocalDisk { } } +async fn wait_for_startup_cleanup_signal( + startup_cleanup_ready: &AtomicU32, + startup_cleanup_notify: &Notify, + wait_timeout: Duration, +) -> bool { + if startup_cleanup_ready.load(Ordering::Acquire) != 0 { + return true; + } + + timeout(wait_timeout, async { + loop { + if startup_cleanup_ready.load(Ordering::Acquire) != 0 { + return; + } + let notified = startup_cleanup_notify.notified(); + if startup_cleanup_ready.load(Ordering::Acquire) != 0 { + return; + } + notified.await; + } + }) + .await + .is_ok() +} + #[tracing::instrument] async fn get_disk_info(drive_path: PathBuf) -> Result<(rustfs_utils::os::DiskInfo, bool)> { let drive_path = drive_path.to_string_lossy().to_string(); @@ -3157,7 +3227,9 @@ mod test { fs::create_dir_all(leftover.parent().unwrap()).await.unwrap(); fs::write(&leftover, b"temporary").await.unwrap(); - LocalDisk::cleanup_tmp_on_startup(dir.path()).await.unwrap(); + LocalDisk::cleanup_tmp_on_startup(dir.path(), Arc::new(AtomicU32::new(0)), Arc::new(Notify::new())) + .await + .unwrap(); assert!(!tmp.join("leftover").exists()); assert!(LocalDisk::meta_path(dir.path(), RUSTFS_META_TMP_DELETED_BUCKET).exists()); @@ -3213,6 +3285,54 @@ mod test { assert!(entries.next_entry().await.unwrap().is_none()); } + #[tokio::test(start_paused = true)] + async fn cleanup_loop_interval_does_not_tick_immediately() { + let start_at = tokio::time::Instant::now() + DELETED_OBJECTS_CLEANUP_INTERVAL; + let mut interval = interval_at(start_at, DELETED_OBJECTS_CLEANUP_INTERVAL); + + assert!(tokio::time::timeout(Duration::from_secs(1), interval.tick()).await.is_err()); + + tokio::time::advance(DELETED_OBJECTS_CLEANUP_INTERVAL).await; + interval.tick().await; + } + + #[tokio::test(start_paused = true)] + async fn startup_cleanup_barrier_waits_for_notification() { + let ready = Arc::new(AtomicU32::new(0)); + let notify = Arc::new(Notify::new()); + + let wait = tokio::spawn({ + let ready = ready.clone(); + let notify = notify.clone(); + async move { wait_for_startup_cleanup_signal(ready.as_ref(), notify.as_ref(), Duration::from_secs(2)).await } + }); + + tokio::task::yield_now().await; + assert!(!wait.is_finished()); + + ready.store(1, Ordering::Release); + notify.notify_waiters(); + + assert!(wait.await.unwrap()); + } + + #[tokio::test(start_paused = true)] + async fn startup_cleanup_barrier_times_out() { + let ready = Arc::new(AtomicU32::new(0)); + let notify = Arc::new(Notify::new()); + + let wait = tokio::spawn({ + let ready = ready.clone(); + let notify = notify.clone(); + async move { wait_for_startup_cleanup_signal(ready.as_ref(), notify.as_ref(), Duration::from_secs(2)).await } + }); + + tokio::task::yield_now().await; + tokio::time::advance(Duration::from_secs(2)).await; + + assert!(!wait.await.unwrap()); + } + #[tokio::test] async fn test_scan_dir_includes_nested_object_dirs() { use rustfs_filemeta::MetacacheReader; diff --git a/crates/object-capacity/src/capacity_manager.rs b/crates/object-capacity/src/capacity_manager.rs index f67d74d18..3104f7da6 100644 --- a/crates/object-capacity/src/capacity_manager.rs +++ b/crates/object-capacity/src/capacity_manager.rs @@ -976,7 +976,7 @@ pub async fn start_background_task(disks: Vec) { } tokio::spawn(async move { - let mut timer = tokio::time::interval(refresh_interval); + let mut timer = tokio::time::interval_at(tokio::time::Instant::now() + refresh_interval, refresh_interval); loop { timer.tick().await; @@ -1002,7 +1002,7 @@ pub async fn start_background_task(disks: Vec) { }); tokio::spawn(async move { - let mut timer = tokio::time::interval(metrics_interval); + let mut timer = tokio::time::interval_at(tokio::time::Instant::now() + metrics_interval, metrics_interval); loop { timer.tick().await; manager_for_metrics.log_runtime_summary().await; diff --git a/rustfs/src/storage/rpc/node_service.rs b/rustfs/src/storage/rpc/node_service.rs index 0e2c85441..6cf1980e6 100644 --- a/rustfs/src/storage/rpc/node_service.rs +++ b/rustfs/src/storage/rpc/node_service.rs @@ -107,11 +107,15 @@ impl Node for NodeService { debug!("PING"); let ping_req = request.into_inner(); - let ping_body = flatbuffers::root::(&ping_req.body); - if let Err(e) = ping_body { - error!("{}", e); + if ping_req.body.is_empty() { + debug!("ping_req received empty body; treating request as liveness probe"); } else { - info!("ping_req:body(flatbuffer): {:?}", ping_body); + let ping_body = flatbuffers::root::(&ping_req.body); + if let Err(e) = ping_body { + warn!("invalid ping request body: {}", e); + } else { + info!("ping_req:body(flatbuffer): {:?}", ping_body); + } } let mut fbb = flatbuffers::FlatBufferBuilder::new(); @@ -992,6 +996,23 @@ mod tests { assert!(!ping_response.body.is_empty()); } + #[tokio::test] + async fn test_ping_with_empty_body() { + let service = create_test_node_service(); + + let request = Request::new(PingRequest { + version: 1, + body: Bytes::new(), + }); + + let response = service.ping(request).await; + assert!(response.is_ok()); + + let ping_response = response.unwrap().into_inner(); + assert_eq!(ping_response.version, 1); + assert!(!ping_response.body.is_empty()); + } + #[tokio::test] async fn test_heal_bucket_invalid_options() { let service = create_test_node_service();