fix(ec): make ec.decode write-path crash-safe and atomic (#9949)

* fix(ec): check decode .idx writes and fsync decoded .dat/.idx

WriteIdxFileFromEcIndex silently dropped io.Copy and Write errors, so a
short or failed write of the reconstructed .idx went unnoticed and the
caller proceeded to delete the source EC shards. Propagate those errors.

Also fsync the decoded .dat and .idx before returning, so the bytes are
durable before the shards that produced them are removed cluster-wide.
Mirror the .idx fsync into the Rust volume server (its .dat already
syncs and its writes already propagate errors).

* fix(ec): publish decoded .dat/.idx atomically via temp file and rename

WriteDatFile and WriteIdxFileFromEcIndex wrote in place at the final
name with O_TRUNC. A crash mid-write left a truncated .dat/.idx at the
final name beside the still-present EC shards; on restart that partial
file could be mounted as the live volume even though the shards held the
real data. Write to a .tmp file, fsync it, then rename into place and
fsync the directory, so the final name is only ever absent or complete.
A failed decode removes its own temp file rather than leaking it.

Add util.FsyncDir as the shared directory-fsync primitive and reuse the
Rust volume server's fsync_dir for the mirrored change.

* fix(ec): propagate .ecj read errors in the Rust decoder

Path::exists returned false for any error (permission denied, transient
IO), silently skipping the deletion journal and resurrecting deleted
needles as live. Read the journal directly and treat only NotFound as
absent, propagating other errors. The Go decoder already behaves this
way (FileExists returns false only for IsNotExist, then the open
surfaces other errors).

* fix(ec): remove rename destination on Windows in the Rust decoder publish

std::fs::rename does not replace an existing file on every Windows
version. Remove the destination first under a Windows guard before the
atomic publish rename, matching the compaction commit path.

seaweed-volume/src/storage/erasure_coding/ec_decoder.rs

@@ -11,7 +11,7 @@ use crate::storage::idx;
 use crate::storage::needle::needle::get_actual_size;
 use crate::storage::super_block::SUPER_BLOCK_SIZE;
 use crate::storage::types::*;
-use crate::storage::volume::volume_file_name;
+use crate::storage::volume::{fsync_dir, volume_file_name};
 
 /// Calculate .dat file size from the max offset entry in .ecx.
 /// Reads the volume version from the first EC shard (.ec00) superblock,
@@ -123,60 +123,81 @@ pub fn write_dat_file_from_shards_with_dirs(
     }
     let base = volume_file_name(dat_dir, collection, volume_id);
     let dat_path = format!("{}.dat", base);
+    // Write to a temp file and atomically rename into place, so a crash
+    // mid-write never leaves a partial .dat at the final name beside the
+    // source shards.
+    let tmp_path = format!("{}.tmp", dat_path);
 
-    // Open data shards from their individual home dirs.
-    let mut shards: Vec<EcVolumeShard> = (0..data_shards as u8)
-        .map(|i| EcVolumeShard::new(&shard_dirs[i as usize], collection, volume_id, i))
-        .collect();
+    let write_result = (|| -> io::Result<()> {
+        // Open data shards from their individual home dirs.
+        let mut shards: Vec<EcVolumeShard> = (0..data_shards as u8)
+            .map(|i| EcVolumeShard::new(&shard_dirs[i as usize], collection, volume_id, i))
+            .collect();
 
-    for shard in &mut shards {
-        shard.open()?;
-    }
-
-    let mut dat_file = File::create(&dat_path)?;
-    let mut remaining = dat_file_size;
-    let large_block_size = ERASURE_CODING_LARGE_BLOCK_SIZE;
-    let small_block_size = ERASURE_CODING_SMALL_BLOCK_SIZE;
-    let large_row_size = (large_block_size * data_shards) as i64;
-
-    let mut shard_offset: u64 = 0;
-
-    // Read large blocks
-    while remaining >= large_row_size {
-        for i in 0..data_shards {
-            let mut buf = vec![0u8; large_block_size];
-            shards[i].read_at(&mut buf, shard_offset)?;
-            let to_write = large_block_size.min(remaining as usize);
-            dat_file.write_all(&buf[..to_write])?;
-            remaining -= to_write as i64;
-            if remaining <= 0 {
-                break;
-            }
+        for shard in &mut shards {
+            shard.open()?;
         }
-        shard_offset += large_block_size as u64;
-    }
 
-    // Read small blocks
-    while remaining > 0 {
-        for i in 0..data_shards {
-            let mut buf = vec![0u8; small_block_size];
-            shards[i].read_at(&mut buf, shard_offset)?;
-            let to_write = small_block_size.min(remaining as usize);
-            dat_file.write_all(&buf[..to_write])?;
-            remaining -= to_write as i64;
-            if remaining <= 0 {
-                break;
+        let mut dat_file = File::create(&tmp_path)?;
+        let mut remaining = dat_file_size;
+        let large_block_size = ERASURE_CODING_LARGE_BLOCK_SIZE;
+        let small_block_size = ERASURE_CODING_SMALL_BLOCK_SIZE;
+        let large_row_size = (large_block_size * data_shards) as i64;
+
+        let mut shard_offset: u64 = 0;
+
+        // Read large blocks
+        while remaining >= large_row_size {
+            for i in 0..data_shards {
+                let mut buf = vec![0u8; large_block_size];
+                shards[i].read_at(&mut buf, shard_offset)?;
+                let to_write = large_block_size.min(remaining as usize);
+                dat_file.write_all(&buf[..to_write])?;
+                remaining -= to_write as i64;
+                if remaining <= 0 {
+                    break;
+                }
             }
+            shard_offset += large_block_size as u64;
         }
-        shard_offset += small_block_size as u64;
-    }
 
-    for shard in &mut shards {
-        shard.close();
-    }
+        // Read small blocks
+        while remaining > 0 {
+            for i in 0..data_shards {
+                let mut buf = vec![0u8; small_block_size];
+                shards[i].read_at(&mut buf, shard_offset)?;
+                let to_write = small_block_size.min(remaining as usize);
+                dat_file.write_all(&buf[..to_write])?;
+                remaining -= to_write as i64;
+                if remaining <= 0 {
+                    break;
+                }
+            }
+            shard_offset += small_block_size as u64;
+        }
 
-    dat_file.sync_all()?;
-    Ok(())
+        for shard in &mut shards {
+            shard.close();
+        }
+
+        // fsync, rename, then fsync the dir so the decoded .dat is durable and
+        // atomically published before the caller deletes the source shards.
+        dat_file.sync_all()?;
+        drop(dat_file);
+        // Windows rename does not replace an existing file on every version;
+        // remove the destination first, matching the compaction commit path.
+        #[cfg(windows)]
+        {
+            let _ = std::fs::remove_file(&dat_path);
+        }
+        std::fs::rename(&tmp_path, &dat_path)?;
+        fsync_dir(&dat_path)?;
+        Ok(())
+    })();
+    if write_result.is_err() {
+        let _ = std::fs::remove_file(&tmp_path);
+    }
+    write_result
 }
 
 /// Write .idx file from .ecx index + .ecj deletion journal.
@@ -192,34 +213,59 @@ pub fn write_idx_file_from_ec_index(
     let ecx_path = format!("{}.ecx", base);
     let ecj_path = format!("{}.ecj", base);
     let idx_path = format!("{}.idx", base);
+    // Write to a temp file and atomically rename into place, so a crash
+    // mid-write never leaves a partial .idx at the final name beside the
+    // source shards.
+    let tmp_path = format!("{}.tmp", idx_path);
 
-    // Copy .ecx to .idx
-    std::fs::copy(&ecx_path, &idx_path)?;
+    let write_result = (|| -> io::Result<()> {
+        // Copy .ecx to the temp .idx
+        std::fs::copy(&ecx_path, &tmp_path)?;
 
-    // Append deletions from .ecj as tombstones
-    if std::path::Path::new(&ecj_path).exists() {
-        let ecj_data = std::fs::read(&ecj_path)?;
-        if !ecj_data.is_empty() {
-            let mut idx_file = std::fs::OpenOptions::new()
-                .write(true)
-                .append(true)
-                .open(&idx_path)?;
-
-            let count = ecj_data.len() / NEEDLE_ID_SIZE;
-            for i in 0..count {
-                let start = i * NEEDLE_ID_SIZE;
-                let needle_id = NeedleId::from_bytes(&ecj_data[start..start + NEEDLE_ID_SIZE]);
-                idx::write_index_entry(
-                    &mut idx_file,
-                    needle_id,
-                    Offset::default(),
-                    TOMBSTONE_FILE_SIZE,
-                )?;
+        // Append deletions from .ecj as tombstones. Read the journal directly
+        // and treat only NotFound as "no journal": Path::exists would also
+        // swallow a permission/IO error and silently skip deletions, which
+        // would resurrect deleted needles as live.
+        let mut idx_file = std::fs::OpenOptions::new()
+            .write(true)
+            .append(true)
+            .open(&tmp_path)?;
+        match std::fs::read(&ecj_path) {
+            Ok(ecj_data) => {
+                let count = ecj_data.len() / NEEDLE_ID_SIZE;
+                for i in 0..count {
+                    let start = i * NEEDLE_ID_SIZE;
+                    let needle_id = NeedleId::from_bytes(&ecj_data[start..start + NEEDLE_ID_SIZE]);
+                    idx::write_index_entry(
+                        &mut idx_file,
+                        needle_id,
+                        Offset::default(),
+                        TOMBSTONE_FILE_SIZE,
+                    )?;
+                }
             }
+            Err(e) if e.kind() == io::ErrorKind::NotFound => {}
+            Err(e) => return Err(e),
         }
-    }
 
-    Ok(())
+        // fsync, rename, then fsync the dir so the decoded .idx is durable and
+        // atomically published before the caller deletes the source shards.
+        idx_file.sync_all()?;
+        drop(idx_file);
+        // Windows rename does not replace an existing file on every version;
+        // remove the destination first, matching the compaction commit path.
+        #[cfg(windows)]
+        {
+            let _ = std::fs::remove_file(&idx_path);
+        }
+        std::fs::rename(&tmp_path, &idx_path)?;
+        fsync_dir(&idx_path)?;
+        Ok(())
+    })();
+    if write_result.is_err() {
+        let _ = std::fs::remove_file(&tmp_path);
+    }
+    write_result
 }
 
 #[cfg(test)]
@@ -288,6 +334,10 @@ mod tests {
             .unwrap();
         write_idx_file_from_ec_index(dir, "", VolumeId(1)).unwrap();
 
+        // Atomic publish must rename the temp files away, never leaving them behind.
+        assert!(!std::path::Path::new(&format!("{}/1.dat.tmp", dir)).exists());
+        assert!(!std::path::Path::new(&format!("{}/1.idx.tmp", dir)).exists());
+
         // Verify reconstructed .dat matches original
         let reconstructed_dat = std::fs::read(format!("{}/1.dat", dir)).unwrap();
         assert_eq!(
@@ -319,4 +369,16 @@ mod tests {
             assert_eq!(&n.data, expected_data, "needle {} data should match", id);
         }
     }
+
+    #[test]
+    fn test_decode_missing_shard_leaves_no_dat() {
+        let tmp = TempDir::new().unwrap();
+        let dir = tmp.path().to_str().unwrap();
+        // No shard files exist, so de-striping must fail and publish nothing:
+        // neither the final .dat nor a partial .dat.tmp may remain.
+        let res = write_dat_file_from_shards(dir, "", VolumeId(7), 100, 10);
+        assert!(res.is_err());
+        assert!(!std::path::Path::new(&format!("{}/7.dat", dir)).exists());
+        assert!(!std::path::Path::new(&format!("{}/7.dat.tmp", dir)).exists());
+    }
 }

seaweed-volume/src/storage/volume.rs

@@ -3419,7 +3419,7 @@ fn get_append_at_ns(last: u64) -> u64 {
 /// proceed with an undurable rename or marker. A path with no openable parent is
 /// tolerated; directory fsync is unsupported on Windows, so it is a no-op there
 /// (matching the Go fsyncDir helper, which ignores that error).
-fn fsync_dir(path: &str) -> io::Result<()> {
+pub(crate) fn fsync_dir(path: &str) -> io::Result<()> {
     #[cfg(windows)]
     {
         let _ = path;

weed/storage/erasure_coding/ec_decoder.go

@@ -4,6 +4,7 @@ import (
 	"fmt"
 	"io"
 	"os"
+	"path/filepath"
 
 	"github.com/seaweedfs/seaweedfs/weed/storage/backend"
 	"github.com/seaweedfs/seaweedfs/weed/storage/idx"
@@ -40,23 +41,53 @@ func WriteIdxFileFromEcIndex(baseFileName string) (err error) {
 	}
 	defer ecxFile.Close()
 
-	idxFile, openErr := os.OpenFile(baseFileName+".idx", os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0644)
+	// Write to a temp file and atomically rename into place, so a crash mid-write
+	// never leaves a partial .idx at the final name beside the source shards.
+	idxFileName := baseFileName + ".idx"
+	tmpFileName := idxFileName + ".tmp"
+	idxFile, openErr := os.OpenFile(tmpFileName, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0644)
 	if openErr != nil {
-		return fmt.Errorf("cannot open %s.idx: %v", baseFileName, openErr)
+		return fmt.Errorf("cannot open %s: %v", tmpFileName, openErr)
 	}
-	defer idxFile.Close()
+	committed := false
+	defer func() {
+		idxFile.Close()
+		if !committed {
+			os.Remove(tmpFileName)
+		}
+	}()
 
-	io.Copy(idxFile, ecxFile)
+	if _, err = io.Copy(idxFile, ecxFile); err != nil {
+		return fmt.Errorf("copy ecx to idx for %s: %v", baseFileName, err)
+	}
 
 	err = iterateEcjFile(baseFileName, func(key types.NeedleId) error {
-
 		bytes := needle_map.ToBytes(key, types.Offset{}, types.TombstoneFileSize)
-		idxFile.Write(bytes)
-
+		if _, writeErr := idxFile.Write(bytes); writeErr != nil {
+			return writeErr
+		}
 		return nil
 	})
+	if err != nil {
+		return err
+	}
 
-	return err
+	// fsync, rename, then fsync the dir so the decoded .idx is durable and
+	// atomically published before the caller deletes the source shards.
+	if err = idxFile.Sync(); err != nil {
+		return fmt.Errorf("sync idx for %s: %v", baseFileName, err)
+	}
+	if err = idxFile.Close(); err != nil {
+		return fmt.Errorf("close idx for %s: %v", baseFileName, err)
+	}
+	if err = os.Rename(tmpFileName, idxFileName); err != nil {
+		return fmt.Errorf("rename idx for %s: %v", baseFileName, err)
+	}
+	if err = util.FsyncDir(filepath.Dir(idxFileName)); err != nil {
+		return fmt.Errorf("fsync dir for %s: %v", baseFileName, err)
+	}
+	committed = true
+	return nil
 }
 
 // FindDatFileSize calculate .dat file size from max offset entry
@@ -175,23 +206,31 @@ func iterateEcjFile(baseFileName string, processNeedleFn func(key types.NeedleId
 // WriteDatFile generates .dat from EC shard files (e.g., .ec00 ~ .ec09 for 10+4)
 func WriteDatFile(baseFileName string, datFileSize int64, shardFileNames []string) error {
 
-	datFile, openErr := os.OpenFile(baseFileName+".dat", os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0644)
+	// Write to a temp file and atomically rename into place, so a crash mid-write
+	// never leaves a partial .dat at the final name beside the source shards.
+	datFileName := baseFileName + ".dat"
+	tmpFileName := datFileName + ".tmp"
+	datFile, openErr := os.OpenFile(tmpFileName, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0644)
 	if openErr != nil {
-		return fmt.Errorf("cannot write volume %s.dat: %v", baseFileName, openErr)
+		return fmt.Errorf("cannot write volume %s: %v", tmpFileName, openErr)
 	}
-	defer datFile.Close()
 
 	// Use the actual number of data shards passed in rather than the global
 	// constant, so the de-striping matches the caller's shard set.
 	dataShards := len(shardFileNames)
 	inputFiles := make([]*os.File, dataShards)
 
+	committed := false
 	defer func() {
+		datFile.Close()
 		for shardId := 0; shardId < dataShards; shardId++ {
 			if inputFiles[shardId] != nil {
 				inputFiles[shardId].Close()
 			}
 		}
+		if !committed {
+			os.Remove(tmpFileName)
+		}
 	}()
 
 	for shardId := 0; shardId < dataShards; shardId++ {
@@ -222,6 +261,21 @@ func WriteDatFile(baseFileName string, datFileSize int64, shardFileNames []strin
 		}
 	}
 
+	// fsync, rename, then fsync the dir so the decoded .dat is durable and
+	// atomically published before the caller deletes the source shards.
+	if err := datFile.Sync(); err != nil {
+		return fmt.Errorf("sync dat for %s: %v", baseFileName, err)
+	}
+	if err := datFile.Close(); err != nil {
+		return fmt.Errorf("close dat for %s: %v", baseFileName, err)
+	}
+	if err := os.Rename(tmpFileName, datFileName); err != nil {
+		return fmt.Errorf("rename dat for %s: %v", baseFileName, err)
+	}
+	if err := util.FsyncDir(filepath.Dir(datFileName)); err != nil {
+		return fmt.Errorf("fsync dir for %s: %v", baseFileName, err)
+	}
+	committed = true
 	return nil
 }
 

weed/storage/erasure_coding/ec_decoder_test.go

@@ -174,6 +174,44 @@ func TestWriteIdxFileFromEcIndex_ProcessesEcjJournal(t *testing.T) {
 	}
 }
 
+// TestDecodeAtomicPublish verifies the decoded .idx/.dat are published via a
+// temp file plus rename: a successful write leaves the final file with no
+// leftover .tmp, and a failed write leaves neither the final file nor a
+// partial .tmp beside the source shards.
+func TestDecodeAtomicPublish(t *testing.T) {
+	dir := t.TempDir()
+	base := filepath.Join(dir, "foo_1")
+
+	// .ecx with one live needle so WriteIdxFileFromEcIndex succeeds.
+	ecxData := makeNeedleMapEntry(types.NeedleId(1), types.ToOffset(64), types.Size(100))
+	if err := os.WriteFile(base+".ecx", ecxData, 0644); err != nil {
+		t.Fatalf("write ecx: %v", err)
+	}
+	if err := erasure_coding.WriteIdxFileFromEcIndex(base); err != nil {
+		t.Fatalf("WriteIdxFileFromEcIndex: %v", err)
+	}
+	if _, err := os.Stat(base + ".idx"); err != nil {
+		t.Fatalf("decoded .idx missing: %v", err)
+	}
+	if _, err := os.Stat(base + ".idx.tmp"); !os.IsNotExist(err) {
+		t.Fatalf("expected no leftover .idx.tmp, stat err=%v", err)
+	}
+
+	// A WriteDatFile pointed at a missing shard must fail and leave neither the
+	// final .dat nor a partial .dat.tmp behind.
+	datBase := filepath.Join(dir, "bar_2")
+	missingShards := []string{filepath.Join(dir, "does_not_exist.ec00")}
+	if err := erasure_coding.WriteDatFile(datBase, 100, missingShards); err == nil {
+		t.Fatalf("expected WriteDatFile to fail on missing shard")
+	}
+	if _, err := os.Stat(datBase + ".dat"); !os.IsNotExist(err) {
+		t.Fatalf("failed decode must not leave a .dat, stat err=%v", err)
+	}
+	if _, err := os.Stat(datBase + ".dat.tmp"); !os.IsNotExist(err) {
+		t.Fatalf("failed decode must not leave a .dat.tmp, stat err=%v", err)
+	}
+}
+
 // TestDecodeWithNonEmptyEcj_AllDeleted verifies the full decode pre-processing
 // when .ecj contains deletions for ALL live entries in .ecx.
 // After RebuildEcxFile merges .ecj into .ecx, HasLiveNeedles must return false

weed/storage/erasure_coding/ec_roundtrip_test.go

@@ -329,6 +329,10 @@ func testDecodeDat(t *testing.T, datSize int64) {
 	err = WriteDatFile(decodedBase, datSize, shardFileNames)
 	require.NoError(t, err, "WriteDatFile")
 
+	// The atomic publish must rename the temp file away, never leaving it behind.
+	_, statErr := os.Stat(decodedBase + ".dat.tmp")
+	require.True(t, os.IsNotExist(statErr), "WriteDatFile must not leave a .dat.tmp")
+
 	// 4. Verify decoded .dat matches original
 	decodedData, err := os.ReadFile(decodedBase + ".dat")
 	require.NoError(t, err)

weed/util/file_util.go

@@ -8,6 +8,7 @@ import (
 	"os"
 	"os/user"
 	"path/filepath"
+	"runtime"
 	"strings"
 	"time"
 
@@ -40,6 +41,24 @@ func GetFileSize(file *os.File) (size int64, err error) {
 	return
 }
 
+// FsyncDir flushes a directory entry so a rename/create/unlink inside it
+// survives a crash. Directory fsync is not supported on every platform
+// (notably Windows), where it is skipped rather than treated as an error.
+func FsyncDir(dir string) error {
+	if runtime.GOOS == "windows" {
+		return nil
+	}
+	d, err := os.Open(dir)
+	if err != nil {
+		return err
+	}
+	defer d.Close()
+	if err = d.Sync(); err != nil && !errors.Is(err, os.ErrInvalid) {
+		return fmt.Errorf("sync dir %s: %v", dir, err)
+	}
+	return nil
+}
+
 func FileExists(filename string) bool {
 
 	_, err := os.Stat(filename)