refactor(blob): add NFS-safe file I/O layer

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)?;
+        let mut buf = Vec::with_capacity(records.len() * INDEX_RECORD_SIZE);
         for r in records {
-                file.write_all(&r.encode())?;
+            buf.extend_from_slice(&r.encode());
         }
-            file.sync_all()?;
-        }
-        std::fs::rename(&temp_path, &self.path)?;
-        Ok(())
+        crate::fs::create_atomic(&self.path, &buf)
     }
 
     /// Delete the bucket file.

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<Mutex<File>> 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 {

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<File> {
+    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<File> {
+    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);
+    }
+}

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;

crates/blob/src/meta.rs

@@ -19,14 +19,7 @@ fn write_bin<T: Serialize>(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(())
 }
 

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<Self> {
-        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<Self> {
-        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<Vec<u8>> {
-        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<Vec<u8>> {
-        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.