diff --git a/seaweed-volume/src/storage/erasure_coding/ec_decoder.rs b/seaweed-volume/src/storage/erasure_coding/ec_decoder.rs index 9e8feb61c..6bcb0ada6 100644 --- a/seaweed-volume/src/storage/erasure_coding/ec_decoder.rs +++ b/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 = (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 = (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()); + } } diff --git a/seaweed-volume/src/storage/volume.rs b/seaweed-volume/src/storage/volume.rs index bd4cd5a3f..09269e37d 100644 --- a/seaweed-volume/src/storage/volume.rs +++ b/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; diff --git a/weed/storage/erasure_coding/ec_decoder.go b/weed/storage/erasure_coding/ec_decoder.go index 10f01ae7d..d6175c215 100644 --- a/weed/storage/erasure_coding/ec_decoder.go +++ b/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 } diff --git a/weed/storage/erasure_coding/ec_decoder_test.go b/weed/storage/erasure_coding/ec_decoder_test.go index 31aeeb3e3..cd639fe9d 100644 --- a/weed/storage/erasure_coding/ec_decoder_test.go +++ b/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 diff --git a/weed/storage/erasure_coding/ec_roundtrip_test.go b/weed/storage/erasure_coding/ec_roundtrip_test.go index 0ba3fc990..46540a182 100644 --- a/weed/storage/erasure_coding/ec_roundtrip_test.go +++ b/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) diff --git a/weed/util/file_util.go b/weed/util/file_util.go index ea28d7d9c..29262b930 100644 --- a/weed/util/file_util.go +++ b/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)