From 507faf3a6a61783839199fd2ba6866ae424d41ba Mon Sep 17 00:00:00 2001 From: Zhengchao An Date: Mon, 24 Aug 2026 16:42:03 +0800 Subject: [PATCH] fix(ci): restore post-merge test gates --- crates/e2e_test/src/compression_test.rs | 2 +- crates/ecstore/src/disk/fs.rs | 156 ++++++++++++------------ crates/ecstore/src/disk/local.rs | 27 +++- 3 files changed, 104 insertions(+), 81 deletions(-) diff --git a/crates/e2e_test/src/compression_test.rs b/crates/e2e_test/src/compression_test.rs index 30b1cdce0..1ed0269ec 100644 --- a/crates/e2e_test/src/compression_test.rs +++ b/crates/e2e_test/src/compression_test.rs @@ -71,7 +71,7 @@ fn find_part_files(temp_dir: &str, bucket: &str, object_key: &str) -> io::Result } fn part_files_total_size(part_files: &[PathBuf]) -> io::Result { - part_files.iter().try_fold(0, |total, path| { + part_files.iter().try_fold(0_u64, |total, path| { let metadata = fs::symlink_metadata(path) .map_err(|error| io::Error::new(error.kind(), format!("failed to stat {}: {error}", path.display())))?; if !metadata.file_type().is_file() { diff --git a/crates/ecstore/src/disk/fs.rs b/crates/ecstore/src/disk/fs.rs index 60e986ecc..585128cf7 100644 --- a/crates/ecstore/src/disk/fs.rs +++ b/crates/ecstore/src/disk/fs.rs @@ -13,9 +13,11 @@ // limitations under the License. use std::{ + collections::HashMap, fs::Metadata, - path::Path, - sync::{Arc, OnceLock}, + path::{Path, PathBuf}, + sync::{Arc, Mutex, OnceLock}, + time::{Duration, Instant}, }; use tokio::{ fs::{self, File}, @@ -225,6 +227,79 @@ pub async fn read_file(path: impl AsRef) -> io::Result> { fs::read(path.as_ref()).await } +// Bucket existence cache - reduces statx syscalls for repeated bucket checks + +/// Cache for bucket directory existence checks. +struct BucketExistenceCache { + cache: Mutex>, + ttl: Duration, +} + +impl BucketExistenceCache { + fn new(ttl: Duration) -> Self { + Self { + cache: Mutex::new(HashMap::new()), + ttl, + } + } + + fn check_exists(&self, path: &PathBuf) -> Option { + let mut cache = self.cache.lock().ok()?; + if let Some((timestamp, exists)) = cache.get(path) { + if timestamp.elapsed() < self.ttl { + return Some(*exists); + } + cache.remove(path); + } + None + } + + fn record(&self, path: PathBuf, exists: bool) { + if let Ok(mut cache) = self.cache.lock() { + cache.insert(path, (Instant::now(), exists)); + } + } + + fn invalidate(&self, path: &PathBuf) { + if let Ok(mut cache) = self.cache.lock() { + cache.remove(path); + } + } +} + +static BUCKET_EXISTENCE_CACHE: std::sync::LazyLock = + std::sync::LazyLock::new(|| BucketExistenceCache::new(Duration::from_secs(60))); + +/// Cached access check - reduces statx syscalls +pub async fn cached_access(path: impl AsRef) -> io::Result<()> { + let path_buf = path.as_ref().to_path_buf(); + + if let Some(exists) = BUCKET_EXISTENCE_CACHE.check_exists(&path_buf) { + if exists { + return Ok(()); + } else { + return Err(io::Error::new(io::ErrorKind::NotFound, "bucket not found (cached)")); + } + } + + let result = fs::metadata(&path_buf).await; + + match &result { + Ok(_) => BUCKET_EXISTENCE_CACHE.record(path_buf, true), + Err(e) if e.kind() == io::ErrorKind::NotFound => { + BUCKET_EXISTENCE_CACHE.record(path_buf, false); + } + _ => {} + } + + result?; + Ok(()) +} + +pub fn invalidate_bucket_cache(path: impl AsRef) { + BUCKET_EXISTENCE_CACHE.invalidate(&path.as_ref().to_path_buf()); +} + #[cfg(test)] mod tests { use super::*; @@ -594,80 +669,3 @@ mod tests { assert!(!same_file(&metadata1, &metadata2)); } } - -// Bucket existence cache - reduces statx syscalls for repeated bucket checks -use std::collections::HashMap; -use std::path::PathBuf; -use std::sync::Mutex; -use std::time::{Duration, Instant}; - -/// Cache for bucket directory existence checks. -struct BucketExistenceCache { - cache: Mutex>, - ttl: Duration, -} - -impl BucketExistenceCache { - fn new(ttl: Duration) -> Self { - Self { - cache: Mutex::new(HashMap::new()), - ttl, - } - } - - fn check_exists(&self, path: &PathBuf) -> Option { - let mut cache = self.cache.lock().ok()?; - if let Some((timestamp, exists)) = cache.get(path) { - if timestamp.elapsed() < self.ttl { - return Some(*exists); - } - cache.remove(path); - } - None - } - - fn record(&self, path: PathBuf, exists: bool) { - if let Ok(mut cache) = self.cache.lock() { - cache.insert(path, (Instant::now(), exists)); - } - } - - fn invalidate(&self, path: &PathBuf) { - if let Ok(mut cache) = self.cache.lock() { - cache.remove(path); - } - } -} - -static BUCKET_EXISTENCE_CACHE: std::sync::LazyLock = - std::sync::LazyLock::new(|| BucketExistenceCache::new(Duration::from_secs(60))); - -/// Cached access check - reduces statx syscalls -pub async fn cached_access(path: impl AsRef) -> io::Result<()> { - let path_buf = path.as_ref().to_path_buf(); - - if let Some(exists) = BUCKET_EXISTENCE_CACHE.check_exists(&path_buf) { - if exists { - return Ok(()); - } else { - return Err(io::Error::new(io::ErrorKind::NotFound, "bucket not found (cached)")); - } - } - - let result = fs::metadata(&path_buf).await; - - match &result { - Ok(_) => BUCKET_EXISTENCE_CACHE.record(path_buf, true), - Err(e) if e.kind() == io::ErrorKind::NotFound => { - BUCKET_EXISTENCE_CACHE.record(path_buf, false); - } - _ => {} - } - - result?; - Ok(()) -} - -pub fn invalidate_bucket_cache(path: impl AsRef) { - BUCKET_EXISTENCE_CACHE.invalidate(&path.as_ref().to_path_buf()); -} diff --git a/crates/ecstore/src/disk/local.rs b/crates/ecstore/src/disk/local.rs index 7e0683b5c..e6fe0cd26 100644 --- a/crates/ecstore/src/disk/local.rs +++ b/crates/ecstore/src/disk/local.rs @@ -9981,7 +9981,9 @@ impl DiskAPI for LocalDisk { let volume_dir = self.io_get_bucket_path(volume)?; - if let Err(e) = cached_access(&volume_dir).await { + // Volume creation is a mutation boundary, so it must observe the live + // filesystem rather than a potentially stale existence-cache entry. + if let Err(e) = access(&volume_dir).await { if e.kind() == ErrorKind::NotFound { os::make_dir_all(&volume_dir, self.io_root()).await?; invalidate_bucket_cache(&volume_dir); @@ -18428,6 +18430,29 @@ mod test { let _ = fs::remove_dir_all(&p).await; } + #[tokio::test] + async fn make_volume_rechecks_stale_positive_existence_cache() { + let root_dir = tempfile::tempdir().expect("temporary disk root should be created"); + let endpoint = Endpoint::try_from(root_dir.path().to_string_lossy().as_ref()).expect("endpoint should parse"); + let disk = LocalDisk::new(&endpoint, false).await.expect("local disk should initialize"); + let volume_dir = disk.io_get_bucket_path("bucket").expect("bucket path should resolve"); + + fs::create_dir_all(&volume_dir) + .await + .expect("bucket directory should be created"); + cached_access(&volume_dir) + .await + .expect("existing bucket should populate the cache"); + fs::remove_dir(&volume_dir) + .await + .expect("bucket directory should be removed outside the cache"); + + disk.make_volume("bucket") + .await + .expect("volume creation should recheck the live filesystem"); + assert!(fs::metadata(volume_dir).await.is_ok(), "volume directory should be recreated"); + } + #[tokio::test] async fn test_delete_volume() { let p = "./testv1";