diff --git a/seaweed-volume/src/storage/disk_location.rs b/seaweed-volume/src/storage/disk_location.rs index ca5ea407c..4ba3e9f68 100644 --- a/seaweed-volume/src/storage/disk_location.rs +++ b/seaweed-volume/src/storage/disk_location.rs @@ -10,7 +10,7 @@ use std::io; use std::sync::atomic::{AtomicBool, AtomicI32, AtomicU64, Ordering}; use std::sync::Arc; -use tracing::{info, warn}; +use tracing::warn; use crate::config::MinFreeSpace; use crate::storage::erasure_coding::ec_shard::{ @@ -106,6 +106,12 @@ impl DiskLocation { fs::create_dir_all(&self.idx_directory)?; } + // Recover any interrupted compaction commit before the volume scan. This + // must run here, not inside the .dat loop: that loop is keyed on .dat + // files and would miss the marker-only or already-renamed-.idx states a + // mid-commit crash can leave behind. + self.reconcile_compact_states(); + // Scan for .dat files let entries = fs::read_dir(&self.directory)?; let mut dat_files: Vec<(String, VolumeId)> = Vec::new(); @@ -156,17 +162,8 @@ impl DiskLocation { } } - // Clean up stale compaction temp files - let cpd_path = format!("{}.cpd", volume_name); - let cpx_path = format!("{}.cpx", idx_name); - if std::path::Path::new(&cpd_path).exists() { - info!(volume_id = vid.0, "removing stale compaction file .cpd"); - let _ = fs::remove_file(&cpd_path); - } - if std::path::Path::new(&cpx_path).exists() { - info!(volume_id = vid.0, "removing stale compaction file .cpx"); - let _ = fs::remove_file(&cpx_path); - } + // Stale .cpd/.cpx temp files were already rolled forward or back by + // the reconcile_compact_states pre-pass above. // Check for an incomplete volume (.note means a VolumeCopy was // interrupted). This runs BELOW the empty-stub sweep and EC @@ -253,6 +250,49 @@ impl DiskLocation { Ok(()) } + /// Directory pre-pass that recovers interrupted compaction commits. Collects + /// every volume id that still has a .cpc commit marker or a leftover + /// .cpd/.cpx temp file across the data and idx directories, then rolls the + /// swap forward (marker present) or back (marker absent) per volume. + fn reconcile_compact_states(&self) { + let mut pending: HashSet<(String, VolumeId)> = HashSet::new(); + let mut collect = |dir: &str| { + if let Ok(entries) = fs::read_dir(dir) { + for entry in entries.flatten() { + let name = entry.file_name().into_string().unwrap_or_default(); + let stem = name + .strip_suffix(".cpc") + .or_else(|| name.strip_suffix(".cpd")) + .or_else(|| name.strip_suffix(".cpx")); + if let Some(stem) = stem { + if let Some(key) = parse_collection_volume_id(stem) { + pending.insert(key); + } + } + } + } + }; + collect(&self.directory); + if self.idx_directory != self.directory { + collect(&self.idx_directory); + } + + for (collection, vid) in pending { + // On a runtime reload (SIGHUP), an already-loaded volume may be + // mid-vacuum: its .cpd/.cpx are live, not crash leftovers, and + // rolling them back would clobber the in-flight compaction. Only + // reconcile vids not currently loaded; genuine startup recovery + // runs before any volume is loaded. + if self.volumes.contains_key(&vid) { + continue; + } + let v = Volume::new_unloaded(&self.directory, &self.idx_directory, &collection, vid); + if let Err(e) = v.reconcile_compact_state() { + warn!(volume_id = vid.0, error = %e, "reconcile interrupted compaction failed"); + } + } + } + /// Reports whether the EC files for (collection, vid) on this disk may be /// deleted to reclaim the local .dat. Returns false (delete) only when that /// provably loses no data; every ambiguity returns true (keep), since the @@ -1149,15 +1189,7 @@ fn parse_volume_filename(filename: &str) -> Option<(String, VolumeId)> { .strip_suffix(".dat") .or_else(|| filename.strip_suffix(".vif")) .or_else(|| filename.strip_suffix(".idx"))?; - if let Some(pos) = stem.rfind('_') { - let collection = &stem[..pos]; - let id_str = &stem[pos + 1..]; - let id: u32 = id_str.parse().ok()?; - Some((collection.to_string(), VolumeId(id))) - } else { - let id: u32 = stem.parse().ok()?; - Some((String::new(), VolumeId(id))) - } + parse_collection_volume_id(stem) } // ============================================================================ diff --git a/seaweed-volume/src/storage/volume.rs b/seaweed-volume/src/storage/volume.rs index c92c19fdb..bd4cd5a3f 100644 --- a/seaweed-volume/src/storage/volume.rs +++ b/seaweed-volume/src/storage/volume.rs @@ -588,6 +588,36 @@ impl Volume { Ok(v) } + /// Build a minimal, unloaded Volume holding only the identity/path fields + /// reconcile_compact_state needs. Used by the disk-location load pre-pass to + /// recover an interrupted commit before the volume itself is loaded. + pub fn new_unloaded(dirname: &str, dir_idx: &str, collection: &str, id: VolumeId) -> Self { + Volume { + id, + dir: dirname.to_string(), + dir_idx: dir_idx.to_string(), + collection: collection.to_string(), + dat_file: None, + remote_dat_file: None, + nm: None, + needle_map_kind: NeedleMapKind::InMemory, + data_file_access_control: Arc::new(DataFileAccessControl::default()), + super_block: SuperBlock::default(), + no_write_or_delete: false, + no_write_can_delete: false, + location_disk_space_low: Arc::new(AtomicBool::new(false)), + last_modified_ts_seconds: 0, + last_append_at_ns: 0, + last_compact_index_offset: 0, + last_compact_revision: 0, + is_compacting: false, + compaction_byte_per_second: 0, + last_io_error: Mutex::new(None), + volume_info: PbVolumeInfo::default(), + has_remote_file: false, + } + } + /// Returns true if the volume is currently being compacted. pub fn is_compacting(&self) -> bool { self.is_compacting @@ -2897,28 +2927,14 @@ impl Volume { self.dat_file = None; self.remote_dat_file = None; - let cpd_path = self.file_name(".cpd"); - let cpx_path = self.file_name(".cpx"); - let dat_path = self.file_name(".dat"); - let idx_path = self.file_name(".idx"); - - // Check that compact files exist - if !Path::new(&cpd_path).exists() || !Path::new(&cpx_path).exists() { - return Err(VolumeError::Io(io::Error::new( - io::ErrorKind::NotFound, - "compact files (.cpd/.cpx) not found", - ))); - } - - // Swap files: .cpd → .dat, .cpx → .idx - fs::rename(&cpd_path, &dat_path)?; - fs::rename(&cpx_path, &idx_path)?; - - // Remove any leveldb/redb index files (rebuilt from .idx on reload) - let ldb_path = self.file_name(".ldb"); - let _ = fs::remove_dir_all(&ldb_path); - let rdb_path = self.file_name(".rdb"); - let _ = fs::remove_file(&rdb_path); + // makeup_diff has fsynced the .cpd/.cpx contents. Persist a durable .cpc + // commit marker BEFORE the renames so the two-rename swap is atomic + // across a crash: a marker on disk means the swap is decided and + // reconcile rolls forward; no marker means roll back. Without it, a + // crash between the two renames leaves a stale .idx that a later vacuum + // compacts to empty. + self.write_compact_commit_marker()?; + self.apply_compact_swap()?; // Reload self.load(true, false, 0, self.version())?; @@ -2926,30 +2942,165 @@ impl Volume { Ok(()) } + /// Write and fsync the .cpc marker, then fsync the directory so the marker + /// survives a crash before apply_compact_swap. + fn write_compact_commit_marker(&self) -> Result<(), VolumeError> { + let marker_path = self.file_name(".cpc"); + let f = OpenOptions::new() + .write(true) + .create(true) + .truncate(true) + .open(&marker_path)?; + f.sync_all()?; + drop(f); + fsync_dir(&marker_path)?; + Ok(()) + } + + /// Perform the durable two-rename compaction commit. Idempotent: run both at + /// the tail of commit and by reconcile rolling forward after a crash. It + /// requires BOTH .cpd/.cpx to be present, so a stale or duplicate commit + /// returns an error without deleting the live .dat/.idx. + fn apply_compact_swap(&self) -> Result<(), VolumeError> { + // Normal commit: both temp files must be present, or a stale/duplicate + // commit could clobber the live .dat/.idx. + if !Path::new(&self.file_name(".cpd")).exists() + || !Path::new(&self.file_name(".cpx")).exists() + { + return Err(VolumeError::Io(io::Error::new( + io::ErrorKind::NotFound, + "compact files (.cpd/.cpx) not found", + ))); + } + self.finish_compact_swap() + } + + /// Renames whichever compaction temp file is still present (.cpd->.dat, + /// .cpx->.idx), fsyncs, drops the stale .ldb/.rdb, then clears the .cpc + /// marker. Tolerates a partial state: a crash after the .dat rename but + /// before the .idx rename leaves only .cpx, which must still be applied -- + /// not abandoned, which would pair a fresh .dat with a stale .idx. Existence + /// is checked robustly so a transient error never silently skips the swap. + fn finish_compact_swap(&self) -> Result<(), VolumeError> { + let exists = |p: String| match fs::metadata(&p) { + Ok(_) => true, + Err(e) if e.kind() == io::ErrorKind::NotFound => false, + Err(_) => true, + }; + let dat_path = self.file_name(".dat"); + let idx_path = self.file_name(".idx"); + let cpd_exists = exists(self.file_name(".cpd")); + let cpx_exists = exists(self.file_name(".cpx")); + + if cpd_exists { + #[cfg(windows)] + { + let _ = fs::remove_file(&dat_path); + } + fs::rename(self.file_name(".cpd"), &dat_path)?; + } + if cpx_exists { + #[cfg(windows)] + { + let _ = fs::remove_file(&idx_path); + } + fs::rename(self.file_name(".cpx"), &idx_path)?; + } + if cpd_exists || cpx_exists { + fsync_dir(&dat_path)?; + if self.dir != self.dir_idx { + fsync_dir(&idx_path)?; + } + let _ = fs::remove_dir_all(self.file_name(".ldb")); + let _ = fs::remove_file(self.file_name(".rdb")); + } + + // Clear the marker last and fsync the dir so a restart does not re-run a + // completed swap. + let marker_path = self.file_name(".cpc"); + match fs::remove_file(&marker_path) { + Ok(()) => {} + Err(e) if e.kind() == io::ErrorKind::NotFound => {} + Err(e) => return Err(e.into()), + } + fsync_dir(&marker_path)?; + Ok(()) + } + + /// Recover an interrupted compaction commit on load. When the .cpc marker is + /// present the swap was decided, so roll FORWARD by finishing the renames; + /// when it is absent any leftover .cpd/.cpx are an abandoned generation, so + /// roll BACK by deleting them (and any stale .ldb beside a healthy .idx). + pub fn reconcile_compact_state(&self) -> Result<(), VolumeError> { + let cpc_path = self.file_name(".cpc"); + if Path::new(&cpc_path).exists() { + // Marker present: the swap was decided. Finish whichever rename is + // still pending -- a crash may have completed only the .dat rename, + // so the lone remaining .cpx must still be applied, not abandoned. + // If neither temp file remains, finish_compact_swap clears the marker. + tracing::info!( + volume_id = self.id.0, + "rolling forward interrupted compaction commit" + ); + return self.finish_compact_swap(); + } + + // No marker: roll back any orphan compaction temp files. + let mut rolled_back = false; + for ext in &[".cpd", ".cpx"] { + let p = self.file_name(ext); + if Path::new(&p).exists() { + tracing::info!( + volume_id = self.id.0, + "rolling back orphan compaction file {}", + ext + ); + match fs::remove_file(&p) { + Ok(()) => {} + Err(e) if e.kind() == io::ErrorKind::NotFound => {} + Err(e) => return Err(e.into()), + } + rolled_back = true; + } + } + if rolled_back { + let _ = fs::remove_dir_all(self.file_name(".ldb")); + let _ = fs::remove_file(self.file_name(".rdb")); + } + Ok(()) + } + /// Clean up leftover compaction files (.cpd, .cpx). pub fn cleanup_compact(&self) -> Result<(), VolumeError> { + // Refuse to unlink .cpd/.cpx while a .cpc marker exists: those temp files + // are the only inputs reconcile can roll forward to, so removing them + // mid-commit would strand a decided swap. + if Path::new(&self.file_name(".cpc")).exists() { + return Err(VolumeError::Io(io::Error::new( + io::ErrorKind::Other, + format!( + "volume {}: refusing cleanup while commit marker present", + self.id + ), + ))); + } + let cpd_path = self.file_name(".cpd"); let cpx_path = self.file_name(".cpx"); let cpldb_path = self.file_name(".cpldb"); + let cpc_path = self.file_name(".cpc"); let e1 = fs::remove_file(&cpd_path); let e2 = fs::remove_file(&cpx_path); let e3 = fs::remove_dir_all(&cpldb_path); + let e4 = fs::remove_file(&cpc_path); // Ignore NotFound errors - if let Err(e) = e1 { - if e.kind() != io::ErrorKind::NotFound { - return Err(e.into()); - } - } - if let Err(e) = e2 { - if e.kind() != io::ErrorKind::NotFound { - return Err(e.into()); - } - } - if let Err(e) = e3 { - if e.kind() != io::ErrorKind::NotFound { - return Err(e.into()); + for e in [e1, e2, e3, e4] { + if let Err(e) = e { + if e.kind() != io::ErrorKind::NotFound { + return Err(e.into()); + } } } @@ -3263,9 +3414,31 @@ fn get_append_at_ns(last: u64) -> u64 { /// Remove all files associated with a volume. /// .dat/.idx removals log at info level so destructive calls are traceable. +/// fsync the parent directory of `path` so a rename/create/unlink inside it is +/// durable, propagating a sync failure so the commit path can abort rather than +/// 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<()> { + #[cfg(windows)] + { + let _ = path; + Ok(()) + } + #[cfg(not(windows))] + { + if let Some(parent) = Path::new(path).parent() { + if let Ok(d) = File::open(parent) { + return d.sync_all(); + } + } + Ok(()) + } +} + pub(crate) fn remove_volume_files(base: &str, keep_vif: bool) { for ext in &[ - ".dat", ".idx", ".vif", ".sdx", ".cpd", ".cpx", ".note", ".rdb", + ".dat", ".idx", ".vif", ".sdx", ".cpd", ".cpx", ".cpc", ".note", ".rdb", ] { if *ext == ".vif" && keep_vif { continue; @@ -4814,4 +4987,145 @@ mod tests { ".ecx is an EC sidecar, never touched here" ); } + + /// A crash after the .idx/.dat were consumed but before the .cpx->.idx + /// rename committed leaves only .cpd + .cpx + .cpc. reconcile_compact_state + /// must roll the swap FORWARD and produce a loadable, writable volume + /// holding the compacted needle count. + #[test] + fn test_reconcile_roll_forward_marker_only() { + let tmp = TempDir::new().unwrap(); + let dir = tmp.path().to_str().unwrap(); + let mut v = make_test_volume(dir); + + for i in 1..=6u64 { + let mut n = Needle { + id: NeedleId(i), + cookie: Cookie(i as u32), + data: format!("data-{}", i).into_bytes(), + data_size: format!("data-{}", i).len() as u32, + ..Needle::default() + }; + v.write_needle(&mut n, true).unwrap(); + } + for id in [2u64, 5u64] { + let mut del = Needle { + id: NeedleId(id), + cookie: Cookie(id as u32), + ..Needle::default() + }; + v.delete_needle(&mut del).unwrap(); + } + let expected_live = v.file_count() - v.deleted_count(); + + v.compact_by_index(0, 0, |_| true).unwrap(); + v.close(); + + let cpd = v.file_name(".cpd"); + let cpx = v.file_name(".cpx"); + let cpc = v.file_name(".cpc"); + let dat = v.file_name(".dat"); + let idx = v.file_name(".idx"); + assert!(Path::new(&cpd).exists()); + assert!(Path::new(&cpx).exists()); + + // Simulate the mid-commit crash: original .dat/.idx are gone and only + // the compacted temp files plus the commit marker survive. + fs::remove_file(&dat).unwrap(); + fs::remove_file(&idx).unwrap(); + let _ = fs::remove_dir_all(v.file_name(".ldb")); + fs::write(&cpc, b"").unwrap(); + + v.reconcile_compact_state().unwrap(); + + assert!(!Path::new(&cpd).exists(), ".cpd consumed by roll-forward"); + assert!(!Path::new(&cpx).exists(), ".cpx consumed by roll-forward"); + assert!(!Path::new(&cpc).exists(), ".cpc cleared after swap"); + assert!(Path::new(&dat).exists(), ".dat restored by roll-forward"); + assert!(Path::new(&idx).exists(), ".idx restored by roll-forward"); + + // The rolled-forward files must load as a writable volume with the + // compacted needle count. + let reloaded = make_test_volume(dir); + assert!(!reloaded.is_read_only(), "rolled-forward volume read-only"); + assert_eq!( + reloaded.file_count(), + expected_live, + "rolled-forward volume needle count" + ); + } + + /// With no .cpc marker, leftover .cpd/.cpx are an abandoned generation; + /// reconcile must roll BACK by deleting them and leave the live .dat/.idx. + #[test] + fn test_reconcile_roll_back_no_marker() { + let tmp = TempDir::new().unwrap(); + let dir = tmp.path().to_str().unwrap(); + let mut v = make_test_volume(dir); + + for i in 1..=4u64 { + let mut n = Needle { + id: NeedleId(i), + cookie: Cookie(i as u32), + data: format!("data-{}", i).into_bytes(), + data_size: format!("data-{}", i).len() as u32, + ..Needle::default() + }; + v.write_needle(&mut n, true).unwrap(); + } + v.compact_by_index(0, 0, |_| true).unwrap(); + + let cpd = v.file_name(".cpd"); + let cpx = v.file_name(".cpx"); + let dat = v.file_name(".dat"); + let idx = v.file_name(".idx"); + assert!(Path::new(&cpd).exists()); + assert!(Path::new(&cpx).exists()); + // No .cpc marker exists. + + v.reconcile_compact_state().unwrap(); + + assert!(!Path::new(&cpd).exists(), ".cpd rolled back"); + assert!(!Path::new(&cpx).exists(), ".cpx rolled back"); + assert!(Path::new(&dat).exists(), "live .dat kept"); + assert!(Path::new(&idx).exists(), "live .idx kept"); + } + + /// apply_compact_swap must abort without touching the live .dat/.idx when + /// the .cpd/.cpx temp files are missing (a stale or duplicate commit). + #[test] + fn test_apply_compact_swap_missing_temp_files_preserves_live() { + let tmp = TempDir::new().unwrap(); + let dir = tmp.path().to_str().unwrap(); + let mut v = make_test_volume(dir); + + for i in 1..=3u64 { + let mut n = Needle { + id: NeedleId(i), + cookie: Cookie(i as u32), + data: format!("data-{}", i).into_bytes(), + data_size: format!("data-{}", i).len() as u32, + ..Needle::default() + }; + v.write_needle(&mut n, true).unwrap(); + } + v.sync_to_disk().unwrap(); + + let dat = v.file_name(".dat"); + let idx = v.file_name(".idx"); + let dat_len_before = fs::metadata(&dat).unwrap().len(); + + // No .cpd/.cpx present: the swap must refuse. + assert!( + v.apply_compact_swap().is_err(), + "swap should error when .cpd/.cpx are missing" + ); + assert!(Path::new(&dat).exists(), "live .dat preserved"); + assert!(Path::new(&idx).exists(), "live .idx preserved"); + assert_eq!( + fs::metadata(&dat).unwrap().len(), + dat_len_before, + "live .dat unchanged" + ); + } } diff --git a/weed/storage/disk_location.go b/weed/storage/disk_location.go index 38d41d60a..077706207 100644 --- a/weed/storage/disk_location.go +++ b/weed/storage/disk_location.go @@ -355,6 +355,12 @@ func (l *DiskLocation) loadExistingVolumesWithId(needleMapKind NeedleMapKind, ld workerNum = 10 } } + // Recover any interrupted compaction commit before the volume scan. This + // must run here, not inside loadExistingVolume: that loop is keyed on + // .idx/.vif entries and would miss the marker-only or already-renamed-.idx + // states a mid-commit crash can leave behind. + l.reconcileCompactStates() + l.concurrentLoadingVolumes(needleMapKind, workerNum, ldbTimeout, diskId) glog.V(2).Infof("Store started on dir: %s with %d volumes max %d (disk ID: %d)", l.Directory, len(l.volumes), l.MaxVolumeCount, diskId) @@ -363,6 +369,62 @@ func (l *DiskLocation) loadExistingVolumesWithId(needleMapKind NeedleMapKind, ld } +// reconcileCompactStates is the directory pre-pass that recovers interrupted +// compaction commits. It collects every volume id that still has a .cpc commit +// marker or a leftover .cpd/.cpx temp file across the data and idx directories, +// then runs reconcileCompactState per volume to roll the swap forward (marker +// present) or back (marker absent). +func (l *DiskLocation) reconcileCompactStates() { + type volKey struct { + collection string + vid needle.VolumeId + } + pending := make(map[volKey]bool) + collect := func(dir string) { + entries, err := os.ReadDir(dir) + if err != nil { + return + } + for _, entry := range entries { + if entry.IsDir() { + continue + } + name := entry.Name() + if !strings.HasSuffix(name, ".cpc") && !strings.HasSuffix(name, ".cpd") && !strings.HasSuffix(name, ".cpx") { + continue + } + collection, vid, err := parseCollectionVolumeId(name[:len(name)-4]) + if err != nil { + continue + } + pending[volKey{collection, vid}] = true + } + } + collect(l.Directory) + if l.IdxDirectory != l.Directory { + collect(l.IdxDirectory) + } + + for k := range pending { + // On a runtime reload (SIGHUP -> LoadNewVolumes), an already-loaded + // volume may be mid-vacuum: its .cpd/.cpx are live, not crash + // leftovers, and rolling them back would clobber the in-flight + // compaction (and remove a live .ldb). Only reconcile vids that are + // not currently loaded; genuine startup recovery runs before any + // volume is loaded, so the map is empty then. + l.volumesLock.RLock() + _, loaded := l.volumes[k.vid] + l.volumesLock.RUnlock() + if loaded { + continue + } + v := &Volume{dir: l.Directory, dirIdx: l.IdxDirectory, Collection: k.collection, Id: k.vid} + if err := v.reconcileCompactState(); err != nil { + glog.Errorf("volume %d: reconcile interrupted compaction failed: %v", k.vid, err) + } + } +} + func (l *DiskLocation) DeleteCollectionFromDiskLocation(collection string) (e error) { l.volumesLock.Lock() diff --git a/weed/storage/needle_map/memdb.go b/weed/storage/needle_map/memdb.go index e348a42ce..e4e0db346 100644 --- a/weed/storage/needle_map/memdb.go +++ b/weed/storage/needle_map/memdb.go @@ -118,8 +118,14 @@ func (cm *MemDb) SaveToIdx(idxName string) (ret error) { return } defer func() { - idxFile.Sync() - idxFile.Close() + // The .cpx generated here is renamed to .idx at commit, so a discarded + // fsync or close error could leave a partially-written index after a crash. + if syncErr := idxFile.Sync(); syncErr != nil && ret == nil { + ret = syncErr + } + if closeErr := idxFile.Close(); closeErr != nil && ret == nil { + ret = closeErr + } }() return cm.AscendingVisit(func(value NeedleValue) error { diff --git a/weed/storage/needle_map_leveldb.go b/weed/storage/needle_map_leveldb.go index fe4b14953..68a5358cd 100644 --- a/weed/storage/needle_map_leveldb.go +++ b/weed/storage/needle_map_leveldb.go @@ -117,8 +117,16 @@ func generateLevelDbFile(dbFileName string, indexFile *os.File) error { glog.Fatalf("stat file %s: %v", indexFile.Name(), err) return err } else { - if watermark*NeedleMapEntrySize > uint64(stat.Size()) { - glog.Warningf("wrong watermark %d for filesize %d", watermark, stat.Size()) + // A watermark past the end of the .idx means the .ldb is stale relative + // to the index it must mirror (e.g. an interrupted compaction left the + // old .ldb beside a freshly swapped, shorter .idx). Trusting it would + // replay zero entries and silently poison the needle map, so rebuild + // from offset 0 instead. + // Compare in entries, not bytes: watermark*NeedleMapEntrySize can + // overflow uint64 for a corrupted watermark and wrap past the size check. + if watermark > uint64(stat.Size())/NeedleMapEntrySize { + glog.Warningf("stale watermark %d for %s (filesize %d); rebuilding leveldb from start", watermark, dbFileName, stat.Size()) + watermark = 0 } glog.V(1).Infof("generateLevelDbFile %s, watermark %d, num of entries:%d", dbFileName, watermark, (uint64(stat.Size())-watermark*NeedleMapEntrySize)/NeedleMapEntrySize) } diff --git a/weed/storage/needle_map_leveldb_test.go b/weed/storage/needle_map_leveldb_test.go index 443bc1b3e..798f7a0d6 100644 --- a/weed/storage/needle_map_leveldb_test.go +++ b/weed/storage/needle_map_leveldb_test.go @@ -7,10 +7,70 @@ import ( "sync" "testing" + "github.com/syndtr/goleveldb/leveldb" + "github.com/seaweedfs/seaweedfs/weed/storage/needle" + "github.com/seaweedfs/seaweedfs/weed/storage/needle_map" "github.com/seaweedfs/seaweedfs/weed/storage/types" ) +// A mid-commit crash can leave an old .ldb (with a high watermark) beside a +// freshly swapped, shorter .idx. generateLevelDbFile must distrust the stale +// watermark and rebuild from offset 0 instead of walking past EOF and +// replaying zero entries, which would leave the needle map empty and make +// every live needle a phantom 404. +func TestGenerateLevelDbFileStaleWatermarkRebuilds(t *testing.T) { + dir := t.TempDir() + + idxPath := filepath.Join(dir, "1.idx") + idxFile, err := os.OpenFile(idxPath, os.O_RDWR|os.O_CREATE, 0644) + if err != nil { + t.Fatalf("create idx: %v", err) + } + defer idxFile.Close() + + // A short .idx: three live needles. + const liveCount = 3 + for i := uint64(1); i <= liveCount; i++ { + entry := needle_map.ToBytes(types.Uint64ToNeedleId(i), types.ToOffset(int64(i*1024)), types.Size(512)) + if _, err := idxFile.Write(entry); err != nil { + t.Fatalf("write idx entry: %v", err) + } + } + if err := idxFile.Sync(); err != nil { + t.Fatalf("sync idx: %v", err) + } + + // Seed an .ldb whose stored watermark sits far past the short .idx, mimicking + // the leftover db from a pre-crash, much larger index. + dbPath := filepath.Join(dir, "1.ldb") + db, err := leveldb.OpenFile(dbPath, nil) + if err != nil { + t.Fatalf("open ldb: %v", err) + } + if err := setWatermark(db, watermarkBatchSize); err != nil { + t.Fatalf("set stale watermark: %v", err) + } + db.Close() + + if err := generateLevelDbFile(dbPath, idxFile); err != nil { + t.Fatalf("generateLevelDbFile: %v", err) + } + + db, err = leveldb.OpenFile(dbPath, nil) + if err != nil { + t.Fatalf("reopen ldb: %v", err) + } + defer db.Close() + for i := uint64(1); i <= liveCount; i++ { + keyBytes := make([]byte, types.NeedleIdSize) + types.NeedleIdToBytes(keyBytes, types.Uint64ToNeedleId(i)) + if _, err := db.Get(keyBytes, nil); err != nil { + t.Fatalf("needle %d missing after rebuild (stale watermark poisoned the map): %v", i, err) + } + } +} + func TestLevelDbNeedleMap_Concurrency(t *testing.T) { dir, err := os.MkdirTemp("", "test_leveldb_concurrency") if err != nil { diff --git a/weed/storage/volume_vacuum.go b/weed/storage/volume_vacuum.go index 818bf08f9..94ae586a9 100644 --- a/weed/storage/volume_vacuum.go +++ b/weed/storage/volume_vacuum.go @@ -5,6 +5,7 @@ import ( "fmt" "io" "os" + "path/filepath" "runtime" "time" @@ -198,31 +199,22 @@ func (v *Volume) CommitCompact() error { return e } } else { - if runtime.GOOS == "windows" { - e = os.RemoveAll(v.FileName(".dat")) - if e != nil { - return e - } - e = os.RemoveAll(v.FileName(".idx")) - if e != nil { - return e - } + // makeupDiff has fsynced the .cpd/.cpx contents. Persist a durable .cpc + // commit marker BEFORE renaming so the two renames are atomic across a + // crash: a marker on disk means the swap is decided and reconcile rolls + // forward; no marker means roll back. Without it, a crash between the + // two renames leaves a stale .idx that a later vacuum compacts to empty. + if e = v.writeCompactCommitMarker(); e != nil { + return e } - var e error - if e = os.Rename(v.FileName(".cpd"), v.FileName(".dat")); e != nil { - return fmt.Errorf("rename %s: %v", v.FileName(".cpd"), e) - } - if e = os.Rename(v.FileName(".cpx"), v.FileName(".idx")); e != nil { - return fmt.Errorf("rename %s: %v", v.FileName(".cpx"), e) + if e = v.applyCompactSwap(); e != nil { + return e } } //glog.V(3).Infof("Pretending to be vacuuming...") //time.Sleep(20 * time.Second) - os.RemoveAll(v.FileName(".ldb")) - os.Remove(v.FileName(".rdb")) - glog.V(3).Infof("Loading volume %d commit file...", v.Id) if e = v.load(true, false, v.needleMapKind, 0, v.Version()); e != nil { return e @@ -231,12 +223,144 @@ func (v *Volume) CommitCompact() error { return nil } +// writeCompactCommitMarker writes and fsyncs the .cpc marker, then fsyncs the +// directory so the marker's existence survives a crash before applyCompactSwap. +func (v *Volume) writeCompactCommitMarker() error { + markerPath := v.FileName(".cpc") + f, err := os.OpenFile(markerPath, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0644) + if err != nil { + return fmt.Errorf("create commit marker %s: %v", markerPath, err) + } + if err = f.Sync(); err != nil { + f.Close() + return fmt.Errorf("sync commit marker %s: %v", markerPath, err) + } + if err = f.Close(); err != nil { + return fmt.Errorf("close commit marker %s: %v", markerPath, err) + } + return fsyncDir(filepath.Dir(markerPath)) +} + +// applyCompactSwap performs the durable two-rename compaction commit. It is +// idempotent and is run both at the tail of CommitCompact and by +// reconcileCompactState rolling forward after a crash. It requires the .cpc +// marker and BOTH .cpd/.cpx to be present, so a stale or duplicate commit +// returns an error without deleting the live .dat/.idx. +func (v *Volume) applyCompactSwap() error { + // Normal commit: both temp files must be present, or a stale/duplicate + // commit could clobber the live .dat/.idx. + if !util.FileExists(v.FileName(".cpd")) || !util.FileExists(v.FileName(".cpx")) { + return fmt.Errorf("volume %d compact swap aborted: missing .cpd/.cpx", v.Id) + } + return v.finishCompactSwap() +} + +// finishCompactSwap renames whichever compaction temp file is still present +// (.cpd->.dat, .cpx->.idx), fsyncs, drops the stale .ldb/.rdb, then clears the +// .cpc marker. It tolerates a partial state: a crash after the .dat rename but +// before the .idx rename leaves only .cpx, which must still be applied -- not +// abandoned, which would pair a fresh .dat with a stale .idx. +func (v *Volume) finishCompactSwap() error { + cpdExists := util.FileExists(v.FileName(".cpd")) + cpxExists := util.FileExists(v.FileName(".cpx")) + + if cpdExists { + if runtime.GOOS == "windows" { + if e := os.RemoveAll(v.FileName(".dat")); e != nil { + return e + } + } + if e := os.Rename(v.FileName(".cpd"), v.FileName(".dat")); e != nil { + return fmt.Errorf("rename %s: %v", v.FileName(".cpd"), e) + } + } + if cpxExists { + if runtime.GOOS == "windows" { + if e := os.RemoveAll(v.FileName(".idx")); e != nil { + return e + } + } + if e := os.Rename(v.FileName(".cpx"), v.FileName(".idx")); e != nil { + return fmt.Errorf("rename %s: %v", v.FileName(".cpx"), e) + } + } + if cpdExists || cpxExists { + if e := fsyncDir(filepath.Dir(v.FileName(".dat"))); e != nil { + return e + } + if v.dir != v.dirIdx { + if e := fsyncDir(filepath.Dir(v.FileName(".idx"))); e != nil { + return e + } + } + // A stale .ldb/.rdb mirrors the old .idx; remove it so it can never + // poison the needle map built from the freshly renamed .idx. + os.RemoveAll(v.FileName(".ldb")) + os.Remove(v.FileName(".rdb")) + } + + // Clear the marker last and fsync the dir so a restart does not re-run a + // completed swap. + if e := os.Remove(v.FileName(".cpc")); e != nil && !os.IsNotExist(e) { + return e + } + return fsyncDir(filepath.Dir(v.FileName(".cpc"))) +} + +// reconcileCompactState recovers an interrupted compaction commit on load. When +// the .cpc marker is present the swap was decided, so roll FORWARD by finishing +// the renames; when it is absent any leftover .cpd/.cpx are an abandoned +// generation, so roll BACK by deleting them (and any stale .ldb left next to a +// healthy .idx). It is keyed only on .cpc/.cpd existence so a crash that left +// just the marker, or an already-renamed .idx, is still handled. +func (v *Volume) reconcileCompactState() error { + cpcPath := v.FileName(".cpc") + if util.FileExists(cpcPath) { + // Marker present: the swap was decided. Finish whichever rename is + // still pending -- a crash may have completed only the .dat rename, so + // the lone remaining .cpx must still be applied, not abandoned. If + // neither temp file remains, finishCompactSwap just clears the marker. + glog.V(0).Infof("volume %d: rolling forward interrupted compaction commit", v.Id) + return v.finishCompactSwap() + } + + // No marker: roll back any orphan compaction temp files. + rolledBack := false + for _, ext := range []string{".cpd", ".cpx"} { + p := v.FileName(ext) + if util.FileExists(p) { + glog.V(0).Infof("volume %d: rolling back orphan compaction file %s", v.Id, ext) + if e := os.Remove(p); e != nil && !os.IsNotExist(e) { + return e + } + rolledBack = true + } + } + if rolledBack { + // A stale .ldb may mirror an .idx that never got swapped; drop it so the + // reload rebuilds the needle map from the surviving .idx. + os.RemoveAll(v.FileName(".ldb")) + os.Remove(v.FileName(".rdb")) + } + return nil +} + func (v *Volume) cleanupCompact() error { glog.V(0).Infof("Cleaning up volume %d vacuuming...", v.Id) + // Serialize with CommitCompact's swap and refuse to unlink .cpd/.cpx while a + // .cpc marker exists: those temp files are the only inputs reconcile can roll + // forward to, so removing them mid-commit would strand a decided swap. + v.dataFileAccessLock.Lock() + defer v.dataFileAccessLock.Unlock() + if util.FileExists(v.FileName(".cpc")) { + return fmt.Errorf("volume %d: refusing cleanup while commit marker present", v.Id) + } + e1 := os.Remove(v.FileName(".cpd")) e2 := os.Remove(v.FileName(".cpx")) e3 := os.RemoveAll(v.FileName(".cpldb")) + e4 := os.Remove(v.FileName(".cpc")) if e1 != nil && !os.IsNotExist(e1) { return e1 } @@ -246,6 +370,24 @@ func (v *Volume) cleanupCompact() error { if e3 != nil && !os.IsNotExist(e3) { return e3 } + if e4 != nil && !os.IsNotExist(e4) { + return e4 + } + return nil +} + +// fsyncDir fsyncs a directory so a rename/create/unlink inside it is durable. +// A failure to open the directory for sync is non-fatal on platforms that do +// not support it. +func fsyncDir(dir string) error { + d, err := os.Open(dir) + if err != nil { + return nil + } + 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 } @@ -332,7 +474,12 @@ func (v *Volume) makeupDiff(newDatFileName, newIdxFileName, oldDatFileName, oldI } defer func() { - idx.Sync() + // makeupDiff appends new needles/tombstones to the .cpx; its fsync is the + // durability gate that must succeed before CommitCompact writes the .cpc + // marker and swaps the files. + if syncErr := idx.Sync(); syncErr != nil && err == nil { + err = fmt.Errorf("sync idx %s: %v", newIdxFileName, syncErr) + } idx.Close() }() @@ -491,8 +638,13 @@ func (v *Volume) copyDataBasedOnIndexFile(opts *CompactOptions) (err error) { return err } defer func() { - dstDatBackend.Sync() - dstDatBackend.Close() + // DiskFile.Close performs the final fsync, so its error is the durability + // signal for the .cpd contents. Surface it (only when no earlier error is + // already being returned) so a failed flush aborts the compaction instead + // of leaving a half-written .cpd that CommitCompact would rename live. + if closeErr := dstDatBackend.Close(); closeErr != nil && err == nil { + err = fmt.Errorf("close compacted dat %s: %v", opts.destDatPath, closeErr) + } }() oldNm := needle_map.NewMemDb() @@ -614,7 +766,9 @@ func (v *Volume) copyDataBasedOnIndexFile(opts *CompactOptions) (err error) { return err } defer func() { - indexFile.Sync() + if syncErr := indexFile.Sync(); syncErr != nil && err == nil { + err = fmt.Errorf("sync compacted idx %s: %v", opts.destIdxPath, syncErr) + } indexFile.Close() }() if v.tmpNm != nil { diff --git a/weed/storage/volume_vacuum_crash_safe_test.go b/weed/storage/volume_vacuum_crash_safe_test.go new file mode 100644 index 000000000..577397821 --- /dev/null +++ b/weed/storage/volume_vacuum_crash_safe_test.go @@ -0,0 +1,342 @@ +package storage + +import ( + "os" + "path/filepath" + "testing" + + "github.com/seaweedfs/seaweedfs/weed/stats" + "github.com/seaweedfs/seaweedfs/weed/storage/needle" + "github.com/seaweedfs/seaweedfs/weed/storage/super_block" + "github.com/seaweedfs/seaweedfs/weed/util" +) + +// fileExists is a tiny test helper for asserting file presence. +func mustNotExist(t *testing.T, path string) { + t.Helper() + if _, err := os.Stat(path); !os.IsNotExist(err) { + t.Fatalf("expected %s to be absent, got err=%v", path, err) + } +} + +func mustExist(t *testing.T, path string) { + t.Helper() + if _, err := os.Stat(path); err != nil { + t.Fatalf("expected %s to exist: %v", path, err) + } +} + +// A crash after the .idx was already renamed away but before the .cpx->.idx +// rename committed leaves only .cpd + .cpx + .cpc on disk. The directory +// pre-pass must roll the swap FORWARD: finish the renames and produce a +// loadable, writable volume holding the compacted needle count. Before the +// fix there was no marker and the loader simply discarded the orphan temp +// files, losing the entire compacted generation. +func TestReconcileRollForwardMarkerOnly(t *testing.T) { + dir := t.TempDir() + + v, err := NewVolume(dir, dir, "", 1, NeedleMapLevelDb, &super_block.ReplicaPlacement{}, &needle.TTL{}, 0, needle.GetCurrentVersion(), 0, 0) + if err != nil { + t.Fatalf("volume creation: %v", err) + } + + const liveCount = 6 + for i := uint64(1); i <= liveCount; i++ { + if _, _, _, err := v.writeNeedle2(newRandomNeedle(i), true, false); err != nil { + t.Fatalf("write %d: %v", i, err) + } + } + // Delete two so the compacted generation is strictly smaller than the source. + for _, id := range []uint64{2, 5} { + if _, err := v.deleteNeedle2(newEmptyNeedle(id)); err != nil { + t.Fatalf("delete %d: %v", id, err) + } + } + expectedLive := uint64(liveCount - 2) + + // Generate a valid .cpd/.cpx pair, then close the volume. + if err := v.CompactByIndex(nil); err != nil { + t.Fatalf("compact by index: %v", err) + } + v.Close() + + cpd := filepath.Join(dir, "1.cpd") + cpx := filepath.Join(dir, "1.cpx") + cpc := filepath.Join(dir, "1.cpc") + datPath := filepath.Join(dir, "1.dat") + idxPath := filepath.Join(dir, "1.idx") + mustExist(t, cpd) + mustExist(t, cpx) + + // Simulate the mid-commit crash: the original .dat/.idx are gone and only + // the compacted temp files plus the commit marker survive. + if err := os.Remove(datPath); err != nil { + t.Fatalf("remove dat: %v", err) + } + if err := os.Remove(idxPath); err != nil { + t.Fatalf("remove idx: %v", err) + } + os.RemoveAll(filepath.Join(dir, "1.ldb")) + if err := os.WriteFile(cpc, nil, 0644); err != nil { + t.Fatalf("write marker: %v", err) + } + + // The directory pre-pass should roll the commit forward. + location := NewDiskLocation(dir, 10, util.MinFreeSpace{}, dir, "", nil, stats.DefaultDiskIOProbeConfig()) + defer location.Close() + location.loadExistingVolumes(NeedleMapLevelDb, 0) + + mustNotExist(t, cpd) + mustNotExist(t, cpx) + mustNotExist(t, cpc) + mustExist(t, datPath) + mustExist(t, idxPath) + + reloaded, found := location.FindVolume(needle.VolumeId(1)) + if !found { + t.Fatalf("volume not loaded after roll-forward") + } + if reloaded.IsReadOnly() { + t.Fatalf("rolled-forward volume is read-only") + } + if got := reloaded.FileCount(); got != expectedLive { + t.Fatalf("rolled-forward volume has %d needles, want compacted count %d", got, expectedLive) + } +} + +// A crash AFTER .cpd->.dat but BEFORE .cpx->.idx leaves a compacted .dat, a +// stale .idx, and only .cpx + .cpc. The roll-forward must finish the pending +// .cpx->.idx rename, not abandon it -- abandoning pairs the fresh .dat with the +// stale .idx, which is index corruption. +func TestReconcileRollForwardPartialRename(t *testing.T) { + dir := t.TempDir() + + v, err := NewVolume(dir, dir, "", 1, NeedleMapLevelDb, &super_block.ReplicaPlacement{}, &needle.TTL{}, 0, needle.GetCurrentVersion(), 0, 0) + if err != nil { + t.Fatalf("volume creation: %v", err) + } + const liveCount = 6 + for i := uint64(1); i <= liveCount; i++ { + if _, _, _, err := v.writeNeedle2(newRandomNeedle(i), true, false); err != nil { + t.Fatalf("write %d: %v", i, err) + } + } + for _, id := range []uint64{2, 5} { + if _, err := v.deleteNeedle2(newEmptyNeedle(id)); err != nil { + t.Fatalf("delete %d: %v", id, err) + } + } + expectedLive := uint64(liveCount - 2) + if err := v.CompactByIndex(nil); err != nil { + t.Fatalf("compact by index: %v", err) + } + v.Close() + + cpd := filepath.Join(dir, "1.cpd") + cpx := filepath.Join(dir, "1.cpx") + cpc := filepath.Join(dir, "1.cpc") + datPath := filepath.Join(dir, "1.dat") + idxPath := filepath.Join(dir, "1.idx") + + // Crash after the .dat rename committed but before the .idx one: .cpd has + // become .dat, the stale pre-compact .idx survives, .cpx and the marker remain. + os.RemoveAll(filepath.Join(dir, "1.ldb")) + if err := os.Remove(datPath); err != nil { + t.Fatalf("remove stale dat: %v", err) + } + if err := os.Rename(cpd, datPath); err != nil { + t.Fatalf("rename cpd->dat: %v", err) + } + if err := os.WriteFile(cpc, nil, 0644); err != nil { + t.Fatalf("write marker: %v", err) + } + mustNotExist(t, cpd) + mustExist(t, cpx) + mustExist(t, idxPath) + + location := NewDiskLocation(dir, 10, util.MinFreeSpace{}, dir, "", nil, stats.DefaultDiskIOProbeConfig()) + defer location.Close() + location.loadExistingVolumes(NeedleMapLevelDb, 0) + + mustNotExist(t, cpx) + mustNotExist(t, cpc) + mustExist(t, datPath) + mustExist(t, idxPath) + + reloaded, found := location.FindVolume(needle.VolumeId(1)) + if !found { + t.Fatalf("volume not loaded after partial roll-forward") + } + if got := reloaded.FileCount(); got != expectedLive { + t.Fatalf("volume has %d needles, want compacted count %d (pending .cpx->.idx was not rolled forward)", got, expectedLive) + } +} + +// With no .cpc marker, leftover .cpd/.cpx are an abandoned generation. The +// pre-pass must roll BACK by deleting them, leaving the live .dat/.idx intact. +func TestReconcileRollBackNoMarker(t *testing.T) { + dir := t.TempDir() + + v, err := NewVolume(dir, dir, "", 1, NeedleMapInMemory, &super_block.ReplicaPlacement{}, &needle.TTL{}, 0, needle.GetCurrentVersion(), 0, 0) + if err != nil { + t.Fatalf("volume creation: %v", err) + } + for i := uint64(1); i <= 4; i++ { + if _, _, _, err := v.writeNeedle2(newRandomNeedle(i), true, false); err != nil { + t.Fatalf("write %d: %v", i, err) + } + } + if err := v.CompactByIndex(nil); err != nil { + t.Fatalf("compact by index: %v", err) + } + v.Close() + + cpd := filepath.Join(dir, "1.cpd") + cpx := filepath.Join(dir, "1.cpx") + datPath := filepath.Join(dir, "1.dat") + idxPath := filepath.Join(dir, "1.idx") + mustExist(t, cpd) + mustExist(t, cpx) + // No .cpc marker exists. + + location := NewDiskLocation(dir, 10, util.MinFreeSpace{}, dir, "", nil, stats.DefaultDiskIOProbeConfig()) + defer location.Close() + location.loadExistingVolumes(NeedleMapInMemory, 0) + + mustNotExist(t, cpd) + mustNotExist(t, cpx) + mustExist(t, datPath) + mustExist(t, idxPath) + + reloaded, found := location.FindVolume(needle.VolumeId(1)) + if !found { + t.Fatalf("volume not loaded after roll-back") + } + if got := reloaded.FileCount(); got != 4 { + t.Fatalf("rolled-back volume has %d needles, want original 4", got) + } +} + +// A runtime reload (SIGHUP -> LoadNewVolumes -> reconcileCompactStates) must +// NOT roll back an in-flight vacuum of an already-loaded volume: its .cpd/.cpx +// are live, not crash leftovers. The pre-pass only reconciles vids not loaded. +func TestReconcileSkipsLoadedVolumeMidVacuum(t *testing.T) { + dir := t.TempDir() + + v, err := NewVolume(dir, dir, "", 1, NeedleMapInMemory, &super_block.ReplicaPlacement{}, &needle.TTL{}, 0, needle.GetCurrentVersion(), 0, 0) + if err != nil { + t.Fatalf("volume creation: %v", err) + } + for i := uint64(1); i <= 4; i++ { + if _, _, _, err := v.writeNeedle2(newRandomNeedle(i), true, false); err != nil { + t.Fatalf("write %d: %v", i, err) + } + } + v.Close() + + // Load the volume as a running server would (populates l.volumes). + location := NewDiskLocation(dir, 10, util.MinFreeSpace{}, dir, "", nil, stats.DefaultDiskIOProbeConfig()) + defer location.Close() + location.loadExistingVolumes(NeedleMapInMemory, 0) + loaded, found := location.FindVolume(needle.VolumeId(1)) + if !found { + t.Fatalf("volume not loaded") + } + + // In-flight vacuum: .cpd/.cpx written, commit (.cpc) not yet reached. + if err := loaded.CompactByIndex(nil); err != nil { + t.Fatalf("compact by index: %v", err) + } + cpd := filepath.Join(dir, "1.cpd") + cpx := filepath.Join(dir, "1.cpx") + mustExist(t, cpd) + mustExist(t, cpx) + + // The reload pre-pass must leave the loaded volume's in-flight temp files alone. + location.reconcileCompactStates() + mustExist(t, cpd) + mustExist(t, cpx) +} + +// applyCompactSwap must abort without touching the live .dat/.idx when the +// .cpd/.cpx temp files are missing (a stale or duplicate commit). Before the +// fix a late commit on Windows would RemoveAll the .dat/.idx first and then +// fail the rename, destroying the volume. +func TestApplyCompactSwapMissingTempFilesPreservesLive(t *testing.T) { + dir := t.TempDir() + + v, err := NewVolume(dir, dir, "", 1, NeedleMapInMemory, &super_block.ReplicaPlacement{}, &needle.TTL{}, 0, needle.GetCurrentVersion(), 0, 0) + if err != nil { + t.Fatalf("volume creation: %v", err) + } + for i := uint64(1); i <= 3; i++ { + if _, _, _, err := v.writeNeedle2(newRandomNeedle(i), true, false); err != nil { + t.Fatalf("write %d: %v", i, err) + } + } + v.Close() + + datPath := filepath.Join(dir, "1.dat") + idxPath := filepath.Join(dir, "1.idx") + mustExist(t, datPath) + mustExist(t, idxPath) + datBefore, _ := os.Stat(datPath) + + // No .cpd/.cpx present: the swap must refuse. + if err := v.applyCompactSwap(); err == nil { + t.Fatalf("applyCompactSwap should error when .cpd/.cpx are missing") + } + + mustExist(t, datPath) + mustExist(t, idxPath) + if datAfter, _ := os.Stat(datPath); datAfter.Size() != datBefore.Size() { + t.Fatalf("live .dat changed size: before=%d after=%d", datBefore.Size(), datAfter.Size()) + } +} + +// Deleting a volume and recreating the same id must not leave a stale .cpc +// behind for the next load to roll forward against. removeVolumeFiles must +// sweep the marker along with the rest. +func TestDestroyRemovesCommitMarker(t *testing.T) { + dir := t.TempDir() + + v, err := NewVolume(dir, dir, "", 1, NeedleMapInMemory, &super_block.ReplicaPlacement{}, &needle.TTL{}, 0, needle.GetCurrentVersion(), 0, 0) + if err != nil { + t.Fatalf("volume creation: %v", err) + } + for i := uint64(1); i <= 3; i++ { + if _, _, _, err := v.writeNeedle2(newRandomNeedle(i), true, false); err != nil { + t.Fatalf("write %d: %v", i, err) + } + } + + // Strand a commit marker and its temp files as a crashed commit would. + cpc := filepath.Join(dir, "1.cpc") + cpd := filepath.Join(dir, "1.cpd") + cpx := filepath.Join(dir, "1.cpx") + for _, p := range []string{cpc, cpd, cpx} { + if err := os.WriteFile(p, []byte("x"), 0644); err != nil { + t.Fatalf("write %s: %v", p, err) + } + } + + if err := v.Destroy(false, false); err != nil { + t.Fatalf("destroy: %v", err) + } + + mustNotExist(t, cpc) + mustNotExist(t, cpd) + mustNotExist(t, cpx) + + // Recreate the same id and load: there must be no marker to roll forward. + v2, err := NewVolume(dir, dir, "", 1, NeedleMapInMemory, &super_block.ReplicaPlacement{}, &needle.TTL{}, 0, needle.GetCurrentVersion(), 0, 0) + if err != nil { + t.Fatalf("recreate volume: %v", err) + } + defer v2.Close() + + location := NewDiskLocation(dir, 10, util.MinFreeSpace{}, dir, "", nil, stats.DefaultDiskIOProbeConfig()) + defer location.Close() + location.reconcileCompactStates() + mustNotExist(t, cpc) +} diff --git a/weed/storage/volume_write.go b/weed/storage/volume_write.go index 24b1dbb9d..0d632142f 100644 --- a/weed/storage/volume_write.go +++ b/weed/storage/volume_write.go @@ -142,6 +142,8 @@ func removeVolumeFiles(filename string, keepVif bool) { // compaction deleteAndLog("cpd") deleteAndLog("cpx") + // compaction commit marker + deleteAndLog("cpc") // level db index file deleteAndLog("ldb") // redb index file (Rust volume server)