From 2e7abfbd63eb585adcd16cdef67f71f5950fa7f4 Mon Sep 17 00:00:00 2001 From: weisd Date: Mon, 23 Mar 2026 10:42:16 +0800 Subject: [PATCH] fix(ecstore): preserve raw metadata read semantics (#2258) Signed-off-by: houseme Co-authored-by: houseme Co-authored-by: heihutu --- .../src/delete_objects_versioning_test.rs | 14 +- crates/ecstore/src/disk/local.rs | 71 +--- crates/ecstore/src/file_cache.rs | 332 ------------------ crates/ecstore/src/lib.rs | 1 - rustfs/src/app/object_usecase.rs | 6 +- 5 files changed, 11 insertions(+), 413 deletions(-) delete mode 100644 crates/ecstore/src/file_cache.rs diff --git a/crates/e2e_test/src/delete_objects_versioning_test.rs b/crates/e2e_test/src/delete_objects_versioning_test.rs index 88dcaa5d4..41f872370 100644 --- a/crates/e2e_test/src/delete_objects_versioning_test.rs +++ b/crates/e2e_test/src/delete_objects_versioning_test.rs @@ -16,15 +16,13 @@ //! "In a versioned Bucket, DeleteMarkers are not appearing straight after //! a delete_objects is called." //! -//! Root cause: `delete_versions_internal` wrote new xl.meta to disk via -//! `write_all_private` without invalidating the `GlobalFileCache`. Subsequent -//! calls to `read_metadata` returned the stale cached xl.meta (without the -//! delete marker), making `list_object_versions` show the old version as -//! `IsLatest=true` rather than the new delete marker. +//! Root cause: metadata updates could become temporarily invisible to +//! `list_object_versions`, so the old version was still reported as +//! `IsLatest=true` instead of the newly-created delete marker. //! -//! Fix: `write_all_private` now calls `get_global_file_cache().invalidate()` -//! after every successful write, and `rename_data` also invalidates the cache -//! for the destination path after the atomic rename. +//! Fix: metadata write, delete, and rename paths now make the updated +//! `xl.meta` immediately visible, and the old file-cache shortcut has been +//! removed from the read path. #[cfg(test)] mod tests { diff --git a/crates/ecstore/src/disk/local.rs b/crates/ecstore/src/disk/local.rs index ecb334a27..a8bb2123e 100644 --- a/crates/ecstore/src/disk/local.rs +++ b/crates/ecstore/src/disk/local.rs @@ -29,7 +29,6 @@ use crate::disk::{ os::{check_path_length, is_empty_dir, is_root_disk, rename_all}, }; use crate::erasure_coding::bitrot_verify; -use crate::file_cache::{get_global_file_cache, prefetch_metadata_patterns, read_metadata_cached}; use crate::global::{GLOBAL_IsErasureSD, GLOBAL_RootDiskThreshold}; use bytes::Bytes; use parking_lot::RwLock as ParkingLotRwLock; @@ -492,56 +491,6 @@ impl LocalDisk { Ok(results.into_iter().map(|(_, path)| path).collect()) } - // Optimized metadata reading with caching - async fn read_metadata_cached(&self, path: PathBuf) -> Result> { - read_metadata_cached(path).await - } - - // Smart prefetching for related files - async fn read_version_with_prefetch( - &self, - volume: &str, - path: &str, - version_id: &str, - opts: &ReadOptions, - ) -> Result { - let file_path = self.get_object_path(volume, path)?; - - // Async prefetch related files, don't block current read - if let Some(parent) = file_path.parent() { - prefetch_metadata_patterns(parent, &[STORAGE_FORMAT_FILE, "part.1", "part.2", "part.meta"]).await; - } - - // Main read logic - let file_dir = self.get_bucket_path(volume)?; - let (data, _) = self.read_raw(volume, file_dir, file_path, opts.read_data).await?; - - get_file_info( - &data, - volume, - path, - version_id, - FileInfoOpts { - data: opts.read_data, - include_free_versions: false, - }, - ) - .map_err(|_e| DiskError::Unexpected) - } - - // Batch metadata reading for multiple objects - async fn read_metadata_batch(&self, requests: Vec<(String, String)>) -> Result>>> { - let paths: Vec = requests - .iter() - .map(|(bucket, key)| self.get_object_path(bucket, &format!("{}/{}", key, STORAGE_FORMAT_FILE))) - .collect::>>()?; - - let cache = get_global_file_cache(); - let results = cache.get_metadata_batch(paths).await; - - Ok(results.into_iter().map(|r| r.ok()).collect()) - } - // /// Write to the filesystem atomically. // /// This is done by first writing to a temporary location and then moving the file. // pub(crate) async fn prepare_file_write<'a>(&self, path: &'a PathBuf) -> Result> { @@ -865,8 +814,6 @@ impl LocalDisk { rename_all(tmp_file_path, &file_path, volume_dir).await?; - // Invalidate cache after successful write - get_global_file_cache().invalidate(&file_path).await; Ok(()) } @@ -893,7 +840,6 @@ impl LocalDisk { self.write_all_internal(&file_path, InternalBuf::Owned(buf), sync, skip_parent) .await?; - get_global_file_cache().invalidate(&file_path).await; Ok(()) } // write_all_internal do write file @@ -2125,9 +2071,6 @@ impl DiskAPI for LocalDisk { return Err(err); } - get_global_file_cache().invalidate(&src_file_path).await; - get_global_file_cache().invalidate(&dst_file_path).await; - if let Some(src_file_path_parent) = src_file_path.parent() { if src_volume != super::RUSTFS_META_MULTIPART_BUCKET { let _ = remove_std(src_file_path_parent); @@ -2484,7 +2427,7 @@ impl DiskAPI for LocalDisk { file_path.as_path(), Path::new(format!("{path}{SLASH_SEPARATOR}{STORAGE_FORMAT_FILE}").as_str()), ]); - return rename_all(src_path, dst_path, file_path).await; + return rename_all(&src_path, &dst_path, file_path).await; } self.delete_file(&volume_dir, &xl_path, true, false).await @@ -2607,17 +2550,9 @@ impl DiskAPI for LocalDisk { #[tracing::instrument(skip(self))] async fn read_metadata(&self, volume: &str, path: &str) -> Result { - // Try to use cached file content reading for better performance, with safe fallback let file_path = self.get_object_path(volume, path)?; - // let file_path = file_path.join(Path::new(STORAGE_FORMAT_FILE)); - - // First, try the cache - if let Ok(bytes) = get_global_file_cache().get_file_content(file_path.clone()).await { - return Ok(bytes); - } - - // Fallback to direct read if cache fails - let (data, _) = self.read_metadata_with_dmtime(&file_path).await?; + let volume_dir = self.get_bucket_path(volume)?; + let (data, _) = self.read_all_data_with_dmtime(volume, volume_dir, file_path).await?; Ok(data.into()) } } diff --git a/crates/ecstore/src/file_cache.rs b/crates/ecstore/src/file_cache.rs deleted file mode 100644 index 511eef305..000000000 --- a/crates/ecstore/src/file_cache.rs +++ /dev/null @@ -1,332 +0,0 @@ -// Copyright 2024 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. - -//! High-performance file content and metadata caching using moka -//! -//! This module provides optimized caching for file operations to reduce -//! redundant I/O and improve overall system performance. - -use super::disk::error::{Error, Result}; -use bytes::Bytes; -use moka::future::Cache; -use rustfs_filemeta::FileMeta; -use std::path::{Path, PathBuf}; -use std::sync::Arc; -use std::time::Duration; - -pub struct OptimizedFileCache { - // Use moka as high-performance async cache - metadata_cache: Cache>, - file_content_cache: Cache, - // Performance monitoring - cache_hits: std::sync::atomic::AtomicU64, - cache_misses: std::sync::atomic::AtomicU64, -} - -impl OptimizedFileCache { - pub fn new() -> Self { - Self { - metadata_cache: Cache::builder() - .max_capacity(2048) - .time_to_live(Duration::from_secs(300)) // 5 minutes TTL - .time_to_idle(Duration::from_secs(60)) // 1 minute idle - .build(), - - file_content_cache: Cache::builder() - .max_capacity(512) // Smaller file content cache - .time_to_live(Duration::from_secs(120)) - .weigher(|_key: &PathBuf, value: &Bytes| value.len() as u32) - .build(), - - cache_hits: std::sync::atomic::AtomicU64::new(0), - cache_misses: std::sync::atomic::AtomicU64::new(0), - } - } - - pub async fn get_metadata(&self, path: PathBuf) -> Result> { - if let Some(cached) = self.metadata_cache.get(&path).await { - self.cache_hits.fetch_add(1, std::sync::atomic::Ordering::Relaxed); - return Ok(cached); - } - - self.cache_misses.fetch_add(1, std::sync::atomic::Ordering::Relaxed); - - // Cache miss, read file - let data = tokio::fs::read(&path) - .await - .map_err(|e| Error::other(format!("Read metadata failed: {e}")))?; - - let mut meta = FileMeta::default(); - meta.unmarshal_msg(&data)?; - - let arc_meta = Arc::new(meta); - self.metadata_cache.insert(path, arc_meta.clone()).await; - - Ok(arc_meta) - } - - pub async fn get_file_content(&self, path: PathBuf) -> Result { - if let Some(cached) = self.file_content_cache.get(&path).await { - self.cache_hits.fetch_add(1, std::sync::atomic::Ordering::Relaxed); - return Ok(cached); - } - - self.cache_misses.fetch_add(1, std::sync::atomic::Ordering::Relaxed); - - let data = tokio::fs::read(&path) - .await - .map_err(|e| Error::other(format!("Read file failed: {e}")))?; - - let bytes = Bytes::from(data); - self.file_content_cache.insert(path, bytes.clone()).await; - - Ok(bytes) - } - - // Prefetch related files - pub async fn prefetch_related(&self, base_path: &Path, patterns: &[&str]) { - let mut prefetch_tasks = Vec::new(); - - for pattern in patterns { - let path = base_path.join(pattern); - if tokio::fs::metadata(&path).await.is_ok() { - let cache = self.clone(); - let path_clone = path.clone(); - prefetch_tasks.push(async move { - let _ = cache.get_metadata(path_clone).await; - }); - } - } - - // Parallel prefetch, don't wait for completion - if !prefetch_tasks.is_empty() { - tokio::spawn(async move { - futures::future::join_all(prefetch_tasks).await; - }); - } - } - - // Batch metadata reading with deduplication - pub async fn get_metadata_batch( - &self, - paths: Vec, - ) -> Vec, rustfs_filemeta::Error>> { - let mut results = Vec::with_capacity(paths.len()); - let mut cache_futures = Vec::new(); - - // First, attempt to get from cache - for (i, path) in paths.iter().enumerate() { - if let Some(cached) = self.metadata_cache.get(path).await { - results.push((i, Ok(cached))); - self.cache_hits.fetch_add(1, std::sync::atomic::Ordering::Relaxed); - } else { - cache_futures.push((i, path.clone())); - } - } - - // For cache misses, read from filesystem - if !cache_futures.is_empty() { - let mut fs_results = Vec::new(); - - for (i, path) in cache_futures { - self.cache_misses.fetch_add(1, std::sync::atomic::Ordering::Relaxed); - - match tokio::fs::read(&path).await { - Ok(data) => { - let mut meta = FileMeta::default(); - match meta.unmarshal_msg(&data) { - Ok(_) => { - let arc_meta = Arc::new(meta); - self.metadata_cache.insert(path, arc_meta.clone()).await; - fs_results.push((i, Ok(arc_meta))); - } - Err(e) => { - fs_results.push((i, Err(e))); - } - } - } - Err(_e) => { - fs_results.push((i, Err(rustfs_filemeta::Error::Unexpected))); - } - } - } - - results.extend(fs_results); - } - - // Sort results back to original order - results.sort_by_key(|(i, _)| *i); - results.into_iter().map(|(_, result)| result).collect() - } - - // Invalidate cache entries for a path - pub async fn invalidate(&self, path: &Path) { - self.metadata_cache.remove(path).await; - self.file_content_cache.remove(path).await; - } - - // Get cache statistics - pub fn get_stats(&self) -> FileCacheStats { - let hits = self.cache_hits.load(std::sync::atomic::Ordering::Relaxed); - let misses = self.cache_misses.load(std::sync::atomic::Ordering::Relaxed); - let hit_rate = if hits + misses > 0 { - (hits as f64 / (hits + misses) as f64) * 100.0 - } else { - 0.0 - }; - - FileCacheStats { - metadata_cache_size: self.metadata_cache.entry_count(), - content_cache_size: self.file_content_cache.entry_count(), - cache_hits: hits, - cache_misses: misses, - hit_rate, - total_weight: 0, // Simplified for compatibility - } - } - - // Clear all caches - pub async fn clear(&self) { - self.metadata_cache.invalidate_all(); - self.file_content_cache.invalidate_all(); - - // Wait for invalidation to complete - self.metadata_cache.run_pending_tasks().await; - self.file_content_cache.run_pending_tasks().await; - } -} - -impl Clone for OptimizedFileCache { - fn clone(&self) -> Self { - Self { - metadata_cache: self.metadata_cache.clone(), - file_content_cache: self.file_content_cache.clone(), - cache_hits: std::sync::atomic::AtomicU64::new(self.cache_hits.load(std::sync::atomic::Ordering::Relaxed)), - cache_misses: std::sync::atomic::AtomicU64::new(self.cache_misses.load(std::sync::atomic::Ordering::Relaxed)), - } - } -} - -#[derive(Debug)] -pub struct FileCacheStats { - pub metadata_cache_size: u64, - pub content_cache_size: u64, - pub cache_hits: u64, - pub cache_misses: u64, - pub hit_rate: f64, - pub total_weight: u64, -} - -impl Default for OptimizedFileCache { - fn default() -> Self { - Self::new() - } -} - -// Global cache instance -use std::sync::OnceLock; - -static GLOBAL_FILE_CACHE: OnceLock = OnceLock::new(); - -pub fn get_global_file_cache() -> &'static OptimizedFileCache { - GLOBAL_FILE_CACHE.get_or_init(OptimizedFileCache::new) -} - -// Utility functions for common operations -pub async fn read_metadata_cached(path: PathBuf) -> Result> { - get_global_file_cache().get_metadata(path).await -} - -pub async fn read_file_content_cached(path: PathBuf) -> Result { - get_global_file_cache().get_file_content(path).await -} - -pub async fn prefetch_metadata_patterns(base_path: &Path, patterns: &[&str]) { - get_global_file_cache().prefetch_related(base_path, patterns).await; -} - -#[cfg(test)] -mod tests { - use super::*; - use std::io::Write; - use tempfile::tempdir; - - #[tokio::test] - async fn test_file_cache_basic() { - let cache = OptimizedFileCache::new(); - - // Create a temporary file - let dir = tempdir().unwrap(); - let file_path = dir.path().join("test.txt"); - let mut file = std::fs::File::create(&file_path).unwrap(); - writeln!(file, "test content").unwrap(); - drop(file); - - // First read should be cache miss - let content1 = cache.get_file_content(file_path.clone()).await.unwrap(); - assert_eq!(content1, Bytes::from("test content\n")); - - // Second read should be cache hit - let content2 = cache.get_file_content(file_path.clone()).await.unwrap(); - assert_eq!(content2, content1); - - let stats = cache.get_stats(); - assert!(stats.cache_hits > 0); - assert!(stats.cache_misses > 0); - } - - #[tokio::test] - async fn test_metadata_batch_read() { - let cache = OptimizedFileCache::new(); - - // Create test files - let dir = tempdir().unwrap(); - let mut paths = Vec::new(); - - for i in 0..5 { - let file_path = dir.path().join(format!("test_{i}.txt")); - let mut file = std::fs::File::create(&file_path).unwrap(); - writeln!(file, "content {i}").unwrap(); - paths.push(file_path); - } - - // Note: This test would need actual FileMeta files to work properly - // For now, we just test that the function runs without errors - let results = cache.get_metadata_batch(paths).await; - assert_eq!(results.len(), 5); - } - - #[tokio::test] - async fn test_cache_invalidation() { - let cache = OptimizedFileCache::new(); - - let dir = tempdir().unwrap(); - let file_path = dir.path().join("test.txt"); - let mut file = std::fs::File::create(&file_path).unwrap(); - writeln!(file, "test content").unwrap(); - drop(file); - - // Read file to populate cache - let _ = cache.get_file_content(file_path.clone()).await.unwrap(); - - // Invalidate cache - cache.invalidate(&file_path).await; - - // Next read should be cache miss again - let _ = cache.get_file_content(file_path.clone()).await.unwrap(); - - let stats = cache.get_stats(); - assert!(stats.cache_misses >= 2); - } -} diff --git a/crates/ecstore/src/lib.rs b/crates/ecstore/src/lib.rs index ad79d62fa..5bab6c1e3 100644 --- a/crates/ecstore/src/lib.rs +++ b/crates/ecstore/src/lib.rs @@ -28,7 +28,6 @@ pub mod disks_layout; pub mod endpoints; pub mod erasure_coding; pub mod error; -pub mod file_cache; pub mod global; pub mod metrics_realtime; pub mod notification_sys; diff --git a/rustfs/src/app/object_usecase.rs b/rustfs/src/app/object_usecase.rs index 10510da49..b8b9dd185 100644 --- a/rustfs/src/app/object_usecase.rs +++ b/rustfs/src/app/object_usecase.rs @@ -2713,10 +2713,8 @@ impl DefaultObjectUsecase { // Check Object Lock retention before deletion // TODO: Future optimization (separate PR) - If performance becomes critical under high delete load: - // 1. Integrate OptimizedFileCache (file_cache.rs) into the read_version() path - // 2. Or add a lightweight get_object_lock_info() that only fetches retention metadata - // 3. Or use combined get-and-delete in storage layer with retention check callback - // Note: The project has OptimizedFileCache with moka, but get_object_info doesn't use it yet + // 1. Add a lightweight get_object_lock_info() that only fetches retention metadata + // 2. Or use combined get-and-delete in storage layer with retention check callback let get_opts: ObjectOptions = get_opts(&bucket, &key, version_id_clone, None, &req.headers) .await .map_err(ApiError::from)?;