From 339a597e7e03178c3fd90dfc70a7f82f9ace98bc Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Sat, 13 Jun 2026 20:06:24 -0700 Subject: [PATCH] fix(vacuum): crash-safe compaction commit with a durable .cpc marker, fsync-before-rename, and a reload fence (#9944) * storage: make vacuum/compaction commit crash-safe with a durable .cpc marker A crash mid-compaction-commit could lose or corrupt volume data. The two-rename commit (.cpd->.dat, .cpx->.idx) was not atomic, fsync results were discarded before renaming over a healthy .dat, a stale .ldb could poison the needle map, and a duplicate/late commit could delete the live .dat/.idx outright. Introduce a durable .cpc commit marker so the swap is atomic across a crash: - CommitCompact writes and fsyncs the .cpc marker after makeupDiff fsyncs the .cpd/.cpx, then runs applyCompactSwap: an existence-guarded rename of .cpd->.dat and .cpx->.idx, a directory fsync, removal of the stale .ldb/.rdb, and finally removal of the marker. - reconcileCompactState recovers an interrupted commit on load: roll forward (finish the renames) when the marker is present, roll back (delete the orphan .cpd/.cpx) when it is absent. It runs from a directory pre-pass keyed on .cpd/.cpc existence, since the per-volume loader is keyed on .idx/.vif and misses the marker-only and already-renamed-.idx states. - applyCompactSwap verifies BOTH .cpd and .cpx exist before touching the live files, so a stale-state commit (including the Windows RemoveAll-then-rename path) errors without deleting anything. - Error-check the fsyncs that gate the swap: the .cpd close-fsync and .cpx fsync in copyDataBasedOnIndexFile, the makeupDiff .idx fsync, and MemDb.SaveToIdx. - generateLevelDbFile rebuilds from offset 0 when the stored watermark sits past the end of the .idx, instead of replaying zero entries and poisoning the needle map. - removeVolumeFiles and cleanupCompact sweep the .cpc marker; cleanup refuses to unlink the temp files while a marker is present. Mirror the commit-marker, fsync-before-rename, guard, and load/reconcile logic in the Rust volume server. * storage: don't reconcile an already-loaded volume's compaction state on reload reconcileCompactStates runs in loadExistingVolumes, which is re-invoked at runtime on SIGHUP (Store.LoadNewVolumes). For a volume that is already loaded and mid-vacuum, its .cpd/.cpx are live temp files, not crash leftovers -- rolling them back would clobber the in-flight compaction (and remove a live .ldb out from under an open handle). Skip any vid already present in the volume map; genuine startup recovery runs before any volume is loaded, so the map is empty then. Mirrored in the Rust volume server. Also drop the .note keepVif change that crept into this branch; it belongs to the replica-copy/verify workstream and is restored to master's behavior here so the two changes don't collide. * storage: roll a compaction commit forward per-file, not all-or-nothing A crash after the .cpd->.dat rename but before .cpx->.idx leaves .cpd gone, .cpx and .cpc present, and a stale .idx. The roll-forward required BOTH temp files, so it skipped the swap and cleared the marker, pairing the fresh .dat with the stale .idx (index corruption). Finish whichever temp file remains: extract finishCompactSwap to rename .cpd->.dat and/or .cpx->.idx independently; applyCompactSwap keeps the both-present guard for the normal commit. Existence in the Rust mirror is checked robustly so a transient error never skips the swap. * seaweed-volume: propagate directory fsync failures on the compaction commit path fsync_dir dropped every sync_all error, so the commit could proceed with an undurable marker or rename and a later restart could recover the wrong generation. Return the error and check it at the commit call sites (marker write and the swap), matching the Go fsyncDir which already propagates. Directory fsync stays a no-op on Windows, where it is unsupported. * storage: overflow-safe stale-watermark check when rebuilding the leveldb index watermark*NeedleMapEntrySize can overflow uint64 for a corrupted watermark and wrap below the file size, defeating the stale-.ldb guard. Compare in entries (watermark > size/NeedleMapEntrySize) instead, which is equivalent and cannot overflow. LevelDb-backed needle map is Go-only; no Rust mirror. * storage: propagate idxFile.Close error when writing the compacted index SaveToIdx writes the .cpx that is renamed to .idx at commit; a discarded Close error (buffered data not flushed) could leave a partially-written index after a crash. Surface it in the same durability gate as the fsync. --- seaweed-volume/src/storage/disk_location.rs | 74 +++- seaweed-volume/src/storage/volume.rs | 386 ++++++++++++++++-- weed/storage/disk_location.go | 62 +++ weed/storage/needle_map/memdb.go | 10 +- weed/storage/needle_map_leveldb.go | 12 +- weed/storage/needle_map_leveldb_test.go | 60 +++ weed/storage/volume_vacuum.go | 198 ++++++++- weed/storage/volume_vacuum_crash_safe_test.go | 342 ++++++++++++++++ weed/storage/volume_write.go | 2 + 9 files changed, 1063 insertions(+), 83 deletions(-) create mode 100644 weed/storage/volume_vacuum_crash_safe_test.go 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)