fix(ci): restore post-merge test gates

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<u64> {
-    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() {

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<Path>) -> io::Result<Vec<u8>> {
     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<HashMap<PathBuf, (Instant, bool)>>,
+    ttl: Duration,
+}
+
+impl BucketExistenceCache {
+    fn new(ttl: Duration) -> Self {
+        Self {
+            cache: Mutex::new(HashMap::new()),
+            ttl,
+        }
+    }
+
+    fn check_exists(&self, path: &PathBuf) -> Option<bool> {
+        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<BucketExistenceCache> =
+    std::sync::LazyLock::new(|| BucketExistenceCache::new(Duration::from_secs(60)));
+
+/// Cached access check - reduces statx syscalls
+pub async fn cached_access(path: impl AsRef<Path>) -> 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<Path>) {
+    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<HashMap<PathBuf, (Instant, bool)>>,
-    ttl: Duration,
-}
-
-impl BucketExistenceCache {
-    fn new(ttl: Duration) -> Self {
-        Self {
-            cache: Mutex::new(HashMap::new()),
-            ttl,
-        }
-    }
-
-    fn check_exists(&self, path: &PathBuf) -> Option<bool> {
-        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<BucketExistenceCache> =
-    std::sync::LazyLock::new(|| BucketExistenceCache::new(Duration::from_secs(60)));
-
-/// Cached access check - reduces statx syscalls
-pub async fn cached_access(path: impl AsRef<Path>) -> 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<Path>) {
-    BUCKET_EXISTENCE_CACHE.invalidate(&path.as_ref().to_path_buf());
-}

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";