From b0f229618c16ee5bb20f75eb3e9239c68c5c67af Mon Sep 17 00:00:00 2001 From: rustmailer Date: Wed, 24 Jun 2026 00:03:56 +0800 Subject: [PATCH] refactor(blob): add NFS-safe file I/O layer --- crates/blob/src/bucket.rs | 16 ++-- crates/blob/src/file_pool.rs | 3 +- crates/blob/src/fs.rs | 142 +++++++++++++++++++++++++++++++++++ crates/blob/src/lib.rs | 1 + crates/blob/src/meta.rs | 9 +-- crates/blob/src/segment.rs | 21 +++--- 6 files changed, 161 insertions(+), 31 deletions(-) create mode 100644 crates/blob/src/fs.rs diff --git a/crates/blob/src/bucket.rs b/crates/blob/src/bucket.rs index 7acaed6..208346f 100644 --- a/crates/blob/src/bucket.rs +++ b/crates/blob/src/bucket.rs @@ -1,4 +1,4 @@ -use std::fs::{File, OpenOptions}; +use std::fs::OpenOptions; use std::io::Write; use std::path::{Path, PathBuf}; @@ -218,17 +218,13 @@ impl BucketFile { } /// Rewrite the bucket file with a sorted, deduplicated set of records. + /// Uses atomic temp+rename to be safe on NFS. pub fn rewrite(&self, records: &[IndexRecord]) -> Result<()> { - let temp_path = self.path.with_extension("idx.tmp"); - { - let mut file = File::create(&temp_path)?; - for r in records { - file.write_all(&r.encode())?; - } - file.sync_all()?; + let mut buf = Vec::with_capacity(records.len() * INDEX_RECORD_SIZE); + for r in records { + buf.extend_from_slice(&r.encode()); } - std::fs::rename(&temp_path, &self.path)?; - Ok(()) + crate::fs::create_atomic(&self.path, &buf) } /// Delete the bucket file. diff --git a/crates/blob/src/file_pool.rs b/crates/blob/src/file_pool.rs index 229533b..ec8ec5b 100644 --- a/crates/blob/src/file_pool.rs +++ b/crates/blob/src/file_pool.rs @@ -4,6 +4,7 @@ use std::path::Path; use std::sync::{Arc, Mutex}; use crate::error::Result; +use crate::fs as fs_util; /// Simple LRU pool of open file handles, keyed by segment_id. /// Uses Arc> to allow safe concurrent reads from the same segment. @@ -34,7 +35,7 @@ impl FilePool { } // Open new file - let file = Arc::new(Mutex::new(File::open(path)?)); + let file = Arc::new(Mutex::new(fs_util::open_read(path)?)); // Evict oldest if full if entries.len() >= self.max_entries { diff --git a/crates/blob/src/fs.rs b/crates/blob/src/fs.rs new file mode 100644 index 0000000..5e0ffc1 --- /dev/null +++ b/crates/blob/src/fs.rs @@ -0,0 +1,142 @@ +use std::fs::{self, File, OpenOptions}; +use std::io::{self, Write}; +use std::path::Path; +use std::time::Duration; + +use crate::error::Result; + +/// Max retries for transient filesystem errors (NFS ESTALE, CIFS sharing violations, etc.) +const MAX_RETRIES: u32 = 5; +const RETRY_DELAY: Duration = Duration::from_millis(20); + +/// Check if an I/O error is transient (retryable). +fn is_transient(err: &io::Error) -> bool { + use std::io::ErrorKind; + matches!( + err.kind(), + ErrorKind::TimedOut + | ErrorKind::Interrupted + | ErrorKind::WouldBlock + | ErrorKind::UnexpectedEof + ) || err.raw_os_error() == Some(116) // ESTALE on Linux +} + +/// Open an existing file for reading, with retry on transient errors (NFS ESTALE etc.). +pub fn open_read(path: &Path) -> Result { + let mut last_err = None; + for attempt in 0..MAX_RETRIES { + match File::open(path) { + Ok(f) => return Ok(f), + Err(e) if is_transient(&e) => { + last_err = Some(e); + if attempt > 0 { + std::thread::sleep(RETRY_DELAY * attempt); + } + continue; + } + Err(e) => return Err(e.into()), + } + } + Err(crate::error::Error::Io(last_err.unwrap())) +} + +/// Open an existing file for writing, with retry on transient errors. +pub fn open_write(path: &Path) -> Result { + let mut last_err = None; + for attempt in 0..MAX_RETRIES { + match OpenOptions::new().write(true).open(path) { + Ok(f) => return Ok(f), + Err(e) if is_transient(&e) => { + last_err = Some(e); + if attempt > 0 { + std::thread::sleep(RETRY_DELAY * attempt); + } + continue; + } + Err(e) => return Err(e.into()), + } + } + Err(crate::error::Error::Io(last_err.unwrap())) +} + +/// Create a new file atomically: write content to a temp file, fsync, then rename. +/// Avoids `create_new(true)` which is racy on NFS. +pub fn create_atomic(path: &Path, content: &[u8]) -> Result<()> { + let tmp = path.with_extension( + path.extension() + .map(|e| format!("{}.tmp", e.to_string_lossy())) + .unwrap_or_else(|| "tmp".to_string()), + ); + + { + let mut f = File::create(&tmp)?; + f.write_all(content)?; + f.sync_all()?; + } + + fs::rename(&tmp, path)?; + Ok(()) +} + +/// Truncate an existing file to the given size, with retry. +pub fn truncate(path: &Path, size: u64) -> Result<()> { + let f = open_write(path)?; + f.set_len(size)?; + f.sync_all()?; + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + use tempfile::TempDir; + + #[test] + fn test_open_read_existing() { + let dir = TempDir::new().unwrap(); + let path = dir.path().join("test.txt"); + std::fs::write(&path, b"hello").unwrap(); + + let mut f = open_read(&path).unwrap(); + let mut s = String::new(); + std::io::Read::read_to_string(&mut f, &mut s).unwrap(); + assert_eq!(s, "hello"); + } + + #[test] + fn test_open_read_missing() { + let dir = TempDir::new().unwrap(); + let result = open_read(&dir.path().join("nope.txt")); + assert!(result.is_err()); + } + + #[test] + fn test_create_atomic_success() { + let dir = TempDir::new().unwrap(); + let path = dir.path().join("data.bin"); + create_atomic(&path, b"hello world").unwrap(); + + let content = std::fs::read(&path).unwrap(); + assert_eq!(content, b"hello world"); + // Temp file should not exist + assert!(!dir.path().join("data.bin.tmp").exists()); + } + + #[test] + fn test_create_atomic_overwrites() { + let dir = TempDir::new().unwrap(); + let path = dir.path().join("data.bin"); + create_atomic(&path, b"first").unwrap(); + create_atomic(&path, b"second").unwrap(); + assert_eq!(std::fs::read(&path).unwrap(), b"second"); + } + + #[test] + fn test_truncate() { + let dir = TempDir::new().unwrap(); + let path = dir.path().join("trunc.bin"); + std::fs::write(&path, b"1234567890").unwrap(); + truncate(&path, 5).unwrap(); + assert_eq!(std::fs::metadata(&path).unwrap().len(), 5); + } +} diff --git a/crates/blob/src/lib.rs b/crates/blob/src/lib.rs index 46feff4..78b9958 100644 --- a/crates/blob/src/lib.rs +++ b/crates/blob/src/lib.rs @@ -6,6 +6,7 @@ pub mod compress; pub mod engine; pub mod error; pub mod file_pool; +pub mod fs; pub mod gc; pub mod meta; pub mod recovery; diff --git a/crates/blob/src/meta.rs b/crates/blob/src/meta.rs index 750b2c9..6238e68 100644 --- a/crates/blob/src/meta.rs +++ b/crates/blob/src/meta.rs @@ -19,14 +19,7 @@ fn write_bin(path: &Path, value: &T) -> Result<()> { buf.extend_from_slice(&META_VERSION.to_le_bytes()); buf.extend_from_slice(&payload); - let tmp = path.with_extension("bin.tmp"); - { - use std::io::Write; - let mut f = std::fs::File::create(&tmp)?; - f.write_all(&buf)?; - f.sync_all()?; - } - std::fs::rename(&tmp, path)?; + crate::fs::create_atomic(path, &buf)?; Ok(()) } diff --git a/crates/blob/src/segment.rs b/crates/blob/src/segment.rs index 30e4c5a..06906bd 100644 --- a/crates/blob/src/segment.rs +++ b/crates/blob/src/segment.rs @@ -1,10 +1,11 @@ -use std::fs::{self, File, OpenOptions}; +use std::fs::{self, File}; use std::io::{Read, Seek, SeekFrom, Write}; use std::path::{Path, PathBuf}; use std::sync::Mutex; use crate::checksum; use crate::error::{Error, Result}; +use crate::fs as fs_util; use crate::types::{Codec, ENTRY_HEADER_SIZE, ENTRY_MAGIC, SEGMENT_MAX_SIZE}; /// In-memory representation of a stored entry. @@ -60,10 +61,8 @@ pub struct SegmentWriter { impl SegmentWriter { pub fn create(path: PathBuf, id: u32) -> Result { - let file = OpenOptions::new() - .create_new(true) - .write(true) - .open(&path)?; + // Use create+truncate instead of create_new to avoid NFS O_EXCL issues. + let file = File::create(&path)?; Ok(Self { file, path, @@ -73,7 +72,7 @@ impl SegmentWriter { } pub fn open_append(path: PathBuf, id: u32) -> Result { - let mut file = OpenOptions::new().write(true).open(&path)?; + let mut file = fs_util::open_write(&path)?; file.seek(SeekFrom::End(0))?; let bytes_written = file.stream_position()?; Ok(Self { @@ -188,7 +187,7 @@ impl SegmentReader { /// Read a single entry at the given offset. Returns the entry and the offset of the next entry. pub fn read_entry_at(&self, offset: u64) -> Result<(Entry, u64)> { - let mut file = File::open(&self.path)?; + let mut file = fs_util::open_read(&self.path)?; file.seek(SeekFrom::Start(offset))?; // Read magic @@ -359,7 +358,7 @@ impl SegmentReader { /// Read data portion of an entry (for pread-style reads when you already know offset + data_size). pub fn read_data(&self, offset: u64, data_size: u32) -> Result> { - let mut file = File::open(&self.path)?; + let mut file = fs_util::open_read(&self.path)?; // Skip magic(4) + crc32(4) + flags(1) + codec(1) + key(32) + raw_size(4) + data_size(4) = 50 bytes let data_start = offset + ENTRY_HEADER_SIZE as u64; file.seek(SeekFrom::Start(data_start))?; @@ -370,7 +369,7 @@ impl SegmentReader { /// Read the full entry header + data for verification (used by recovery and GC). pub fn read_full_entry(&self, offset: u64, data_size: u32) -> Result> { - let mut file = File::open(&self.path)?; + let mut file = fs_util::open_read(&self.path)?; file.seek(SeekFrom::Start(offset))?; let total = ENTRY_HEADER_SIZE + data_size as usize; let mut buf = vec![0u8; total]; @@ -416,9 +415,7 @@ impl SegmentReader { /// Truncate a segment file to the given size. pub fn truncate_segment(path: &Path, size: u64) -> Result<()> { - let file = OpenOptions::new().write(true).open(path)?; - file.set_len(size)?; - Ok(()) + fs_util::truncate(path, size) } /// Map an Error, converting Io(StorageFull) to DiskFull with path context.