From 0ef026084427a950564c855215dd08fcc95fbce5 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Wed, 11 Nov 2026 08:48:58 -0800 Subject: [PATCH] volume: fail a vacuum step that finds the volume already compacting (#12553) * volume: fail a vacuum step that finds a compaction already running Compact, CompactByIndex and CommitCompact returned success when another compaction held the volume. A second, uncoordinated vacuum (an expired admin job still running, a shell volume.vacuum next to the worker) then went on to commit or clean up the first run's .cpd/.cpx. Return a "volume N is compacting" error instead, and refuse cleanup while a copy is in flight, as the Rust volume server already did. Co-Authored-By: Claude Opus 5.5 * volume: check the compaction files before a commit closes the volume CommitCompact closed the needle map and data file and wrote the .cpc marker before it found .cpd/.cpx missing. A duplicate commit, such as the second of two overlapping vacuums, then left the volume registered with no open files: it dropped out of the heartbeat and looked lost to volume.list and volume.fix.replication until a restart. Check for the files first so the refused commit leaves the volume as it was. Co-Authored-By: Claude Opus 5.5 * volume: run makeupDiff before a commit closes the volume CommitCompact closed the needle map and data file before makeupDiff, so a makeupDiff failure (a compaction revision another commit moved, an I/O error) returned with them closed: the volume stayed registered but missing from the heartbeat until a restart. makeupDiff reads the live files through its own handles while the commit holds the write lock, so close them only once it has succeeded, as the Rust volume server does. Co-Authored-By: Claude Opus 5.5 * volume: log a volume the heartbeat skips because its files are closed A volume left without its needle map or data file vanished from the heartbeat with no log line, so the master saw a replica disappear while its files were still on disk. Warn on each skip, as the deleted-.dat case next to it already does. Co-Authored-By: Claude Opus 5.5 * volume: claim the compaction flag while cleanup removes the temp files Cleanup only read the flag, so a compaction could start between the check and the removal and lose its .cpd/.cpx. Claim it for the cleanup instead, in Go and Rust. Co-Authored-By: Claude Opus 5.5 --------- Co-authored-by: Claude Opus 5.5 --- seaweed-volume/src/server/grpc_server.rs | 2 +- seaweed-volume/src/storage/store.rs | 6 +- seaweed-volume/src/storage/volume.rs | 64 +++++++++-- weed/storage/volume.go | 1 + weed/storage/volume_vacuum.go | 59 ++++++---- weed/storage/volume_vacuum_test.go | 133 +++++++++++++++++++++++ weed/storage/volume_write.go | 2 +- 7 files changed, 230 insertions(+), 37 deletions(-) diff --git a/seaweed-volume/src/server/grpc_server.rs b/seaweed-volume/src/server/grpc_server.rs index c354d05a10..c08f97a380 100644 --- a/seaweed-volume/src/server/grpc_server.rs +++ b/seaweed-volume/src/server/grpc_server.rs @@ -8068,7 +8068,7 @@ fn run_vacuum_compact( .begin_compact_volume(vid, preallocate); let result = match job { Ok(Some(job)) => job.run(progress), - Ok(None) => Ok(()), // already compacting + Ok(None) => Err(crate::storage::volume::compacting_error(vid)), Err(e) => Err(e), }; diff --git a/seaweed-volume/src/storage/store.rs b/seaweed-volume/src/storage/store.rs index 3fe1051124..2ea07010d8 100644 --- a/seaweed-volume/src/storage/store.rs +++ b/seaweed-volume/src/storage/store.rs @@ -1807,7 +1807,7 @@ impl Store { { match self.begin_compact_volume(vid, preallocate)? { Some(job) => job.run(progress_fn), - None => Ok(()), + None => Err(crate::storage::volume::compacting_error(vid)), } } @@ -3671,8 +3671,8 @@ mod tests { assert!(store.cleanup_compact_volume(vid).is_err()); let (_, v) = store.find_volume(vid).unwrap(); assert!(v.relocate_index_to(other.path().to_str().unwrap()).is_err()); - // Go parity: a commit that finds the volume compacting is a no-op. - store.commit_compact_volume(vid).unwrap(); + assert!(store.commit_compact_volume(vid).is_err()); + assert!(store.compact_volume(vid, 0, 0, |_| true).is_err()); let (_, v) = store.find_volume(vid).expect("still mounted"); assert!(v.is_compacting()); diff --git a/seaweed-volume/src/storage/volume.rs b/seaweed-volume/src/storage/volume.rs index 7ffb4c00aa..774d651baf 100644 --- a/seaweed-volume/src/storage/volume.rs +++ b/seaweed-volume/src/storage/volume.rs @@ -1242,6 +1242,10 @@ const RANGE_CHECK_BLOCKS: u64 = 64; const MIN_RANGE_CHECK_BLOCK_SIZE: u64 = 64 * 1024; const MAX_RANGE_CHECK_BLOCK_SIZE: u64 = 1024 * 1024; +pub(crate) fn compacting_error(id: VolumeId) -> VolumeError { + VolumeError::Io(io::Error::other(format!("volume {} is compacting", id))) +} + /// Keeps a needle to 64 blocks, each 64 KiB to 1 MiB, as Go's /// rangeCheckBlockSize. pub(crate) fn range_check_block_size(data_size: u64) -> u64 { @@ -1913,10 +1917,7 @@ impl Volume { } pub(crate) fn compacting_error(&self) -> VolumeError { - VolumeError::Io(io::Error::other(format!( - "volume {} is compacting", - self.id - ))) + compacting_error(self.id) } // ---- File naming (matching Go) ---- @@ -5843,7 +5844,7 @@ impl Volume { { match self.begin_compact_by_index()? { Some(job) => job.run(progress_fn), - None => Ok(()), // already compacting + None => Err(self.compacting_error()), } } @@ -5948,7 +5949,7 @@ impl Volume { return Err(VolumeError::NotFound); } let Some(_claim) = CompactionClaim::try_claim(&self.is_compacting) else { - return Ok(()); // already compacting, silently skip (matches Go) + return Err(self.compacting_error()); }; // The reload would read the remote object through the compacted .idx. if self.has_remote_file() { @@ -5963,6 +5964,8 @@ impl Volume { } fn do_commit_compact(&self) -> Result<(), VolumeError> { + // A duplicate commit must fail before it closes the live files. + self.check_compact_files()?; if let Err(e) = self.makeup_diff() { warn!("makeup_diff failed: {}", e); // Match Go: clean up .cpd/.cpx on makeup_diff failure @@ -6038,8 +6041,13 @@ impl Volume { /// 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. + self.check_compact_files()?; + self.finish_compact_swap() + } + + /// Both temp files of a normal commit must be present, or a stale or + /// duplicate commit could clobber the live .dat/.idx. + fn check_compact_files(&self) -> Result<(), VolumeError> { if !Path::new(&self.file_name(".cpd")).exists() || !Path::new(&self.file_name(".cpx")).exists() { @@ -6048,7 +6056,7 @@ impl Volume { "compact files (.cpd/.cpx) not found", ))); } - self.finish_compact_swap() + Ok(()) } /// Renames whichever compaction temp file is still present (.cpd->.dat, @@ -6148,10 +6156,11 @@ impl Volume { /// Clean up leftover compaction files (.cpd, .cpx). pub fn cleanup_compact(&self) -> Result<(), VolumeError> { - // An in-flight copy is still writing .cpd/.cpx. - if self.is_compacting() { + // An in-flight copy is still writing .cpd/.cpx; claim the volume so + // none starts while the files are removed. + let Some(_claim) = CompactionClaim::try_claim(&self.is_compacting) else { return Err(self.compacting_error()); - } + }; // 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. @@ -12929,4 +12938,35 @@ mod tests { "live .dat unchanged" ); } + + /// A commit that finds no .cpd/.cpx (another vacuum already committed + /// them) must leave the volume loaded and writable. + #[test] + fn test_commit_compact_without_temp_files_keeps_volume_loaded() { + let tmp = TempDir::new().unwrap(); + let dir = tmp.path().to_str().unwrap(); + let v = make_test_volume(dir); + let needle = |i: u64| 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() + }; + for i in 1..=3u64 { + v.write_needle(&mut needle(i), true, false).unwrap(); + } + v.compact_by_index(0, 0, |_| true).unwrap(); + v.commit_compact().unwrap(); + + assert!(v.commit_compact().is_err()); + assert!(!Path::new(&v.file_name(".cpc")).exists()); + v.write_needle(&mut needle(4), true, false).unwrap(); + let mut n1 = Needle { + id: NeedleId(1), + ..Needle::default() + }; + v.read_needle(&mut n1).unwrap(); + assert_eq!(n1.data, b"data-1"); + } } diff --git a/weed/storage/volume.go b/weed/storage/volume.go index bc70947aab..22a6c42edb 100644 --- a/weed/storage/volume.go +++ b/weed/storage/volume.go @@ -521,6 +521,7 @@ func (v *Volume) ToVolumeInformationMessage(into *master_pb.VolumeInformationMes maxFileKey, volumeSize, modTime, fileCount, deletedCount, deletedSize, ok := v.collectStatus() if !ok { + glog.Warningf("Volume %d: needle map or data file is closed - not reporting to master", v.Id) return 0, nil } diff --git a/weed/storage/volume_vacuum.go b/weed/storage/volume_vacuum.go index 12b86982fc..88ff849a8c 100644 --- a/weed/storage/volume_vacuum.go +++ b/weed/storage/volume_vacuum.go @@ -103,8 +103,7 @@ func (v *Volume) CompactByVolumeData(opts *CompactOptions) error { //defer v.accessLock.Unlock() //glog.V(3).Infof("Got Compaction lock...") if !v.isCompactionInProgress.CompareAndSwap(false, true) { - glog.V(0).Infof("Volume %d is already compacting...", v.Id) - return nil + return v.compactingError() } defer v.isCompactionInProgress.Store(false) @@ -148,8 +147,7 @@ func (v *Volume) CompactByIndex(opts *CompactOptions) error { } if !v.isCompactionInProgress.CompareAndSwap(false, true) { - glog.V(0).Infof("Volume %d is already compacting2 ...", v.Id) - return nil + return v.compactingError() } defer v.isCompactionInProgress.Store(false) @@ -186,8 +184,7 @@ func (v *Volume) CommitCompact() error { glog.V(0).Infof("Committing volume %d vacuuming...", v.Id) if !v.isCompactionInProgress.CompareAndSwap(false, true) { - glog.V(0).Infof("Volume %d is already compacting ...", v.Id) - return nil + return v.compactingError() } defer v.isCompactionInProgress.Store(false) @@ -212,18 +209,10 @@ func (v *Volume) CommitCompact() error { } return errTieredVolumeCompact(v.Id) } - if v.nm != nil { - v.nm.Close() - v.nm = nil + // A duplicate commit must fail before it closes the live files. + if err := v.checkCompactFiles(); err != nil { + return err } - if v.DataBackend != nil { - if err := v.DataBackend.Close(); err != nil { - glog.V(0).Infof("failed to close volume %d", v.Id) - } - } - v.DataBackend = nil - stats.VolumeServerVolumeGauge.WithLabelValues(v.Collection, "volume").Dec() - if compactErr := v.makeupDiff(v.FileName(".cpd"), v.FileName(".cpx"), v.FileName(".dat"), v.FileName(".idx")); compactErr != nil { glog.V(0).Infof("makeupDiff in CommitCompact volume %d failed %v", v.Id, compactErr) if e := os.Remove(v.FileName(".cpd")); e != nil && !os.IsNotExist(e) { @@ -237,6 +226,18 @@ func (v *Volume) CommitCompact() error { return compactErr } + if v.nm != nil { + v.nm.Close() + v.nm = nil + } + if v.DataBackend != nil { + if err := v.DataBackend.Close(); err != nil { + glog.V(0).Infof("failed to close volume %d", v.Id) + } + } + v.DataBackend = nil + stats.VolumeServerVolumeGauge.WithLabelValues(v.Collection, "volume").Dec() + // 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 @@ -270,6 +271,10 @@ func (v *Volume) CommitCompact() error { return nil } +func (v *Volume) compactingError() error { + return fmt.Errorf("volume %d is compacting", v.Id) +} + // 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 { @@ -294,12 +299,19 @@ func (v *Volume) writeCompactCommitMarker() error { // 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 err := v.checkCompactFiles(); err != nil { + return err + } + return v.finishCompactSwap() +} + +// checkCompactFiles requires both temp files of a normal commit, or a +// stale/duplicate commit could clobber the live .dat/.idx. +func (v *Volume) checkCompactFiles() error { 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() + return nil } // finishCompactSwap renames whichever compaction temp file is still present @@ -395,6 +407,13 @@ func (v *Volume) reconcileCompactState() error { func (v *Volume) cleanupCompact() error { glog.V(0).Infof("Cleaning up volume %d vacuuming...", v.Id) + // An in-flight copy is still writing .cpd/.cpx; claim the volume so + // none starts while the files are removed. + if !v.isCompactionInProgress.CompareAndSwap(false, true) { + return v.compactingError() + } + defer v.isCompactionInProgress.Store(false) + // 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. diff --git a/weed/storage/volume_vacuum_test.go b/weed/storage/volume_vacuum_test.go index 28c7f9db75..b144d8096b 100644 --- a/weed/storage/volume_vacuum_test.go +++ b/weed/storage/volume_vacuum_test.go @@ -204,6 +204,136 @@ func TestCommitCompactDeletionTailKeepsWritable(t *testing.T) { } } +// A second vacuum must not take a running compaction as its own: reporting +// success would let it commit or delete the other run's .cpd/.cpx. +func TestCompactionInProgressRefusesOtherVacuumSteps(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) + } + defer v.Close() + for i := uint64(1); i <= 3; i++ { + if _, _, _, err := v.writeNeedle2(newRandomNeedle(i), true, false, false); err != nil { + t.Fatalf("write %d: %v", i, err) + } + } + if err := v.CompactByIndex(nil); err != nil { + t.Fatalf("compact: %v", err) + } + + v.isCompactionInProgress.Store(true) + defer v.isCompactionInProgress.Store(false) + if err := v.CompactByIndex(nil); err == nil { + t.Fatal("CompactByIndex succeeded while another compaction is running") + } + if err := v.CompactByVolumeData(nil); err == nil { + t.Fatal("CompactByVolumeData succeeded while another compaction is running") + } + if err := v.CommitCompact(); err == nil { + t.Fatal("CommitCompact succeeded while another compaction is running") + } + if err := v.cleanupCompact(); err == nil { + t.Fatal("cleanupCompact succeeded while another compaction is running") + } + for _, ext := range []string{".cpd", ".cpx"} { + if _, err := os.Stat(v.FileName(ext)); err != nil { + t.Fatalf("%s of the running compaction: %v", ext, err) + } + } + v.isCompactionInProgress.Store(false) + + if err := v.CommitCompact(); err != nil { + t.Fatalf("commit after the compaction finished: %v", err) + } + if v.SuperBlock.CompactionRevision != 1 { + t.Fatalf("compaction revision %d, want 1", v.SuperBlock.CompactionRevision) + } +} + +// A commit that finds no .cpd/.cpx (another vacuum already committed them) +// must leave the volume loaded, or it drops out of the heartbeat until restart. +func TestCommitCompactWithoutTempFilesKeepsVolumeLoaded(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) + } + defer v.Close() + for i := uint64(1); i <= 3; i++ { + if _, _, _, err := v.writeNeedle2(newRandomNeedle(i), true, false, false); err != nil { + t.Fatalf("write %d: %v", i, err) + } + } + if err := v.CompactByIndex(nil); err != nil { + t.Fatalf("compact: %v", err) + } + if err := v.CommitCompact(); err != nil { + t.Fatalf("first commit: %v", err) + } + + if err := v.CommitCompact(); err == nil { + t.Fatal("second commit succeeded without .cpd/.cpx") + } + if _, _, _, _, _, _, ok := v.collectStatus(); !ok { + t.Fatal("volume dropped out of the heartbeat after the refused commit") + } + if _, err := os.Stat(v.FileName(".cpc")); !os.IsNotExist(err) { + t.Fatalf("refused commit left a commit marker: %v", err) + } + if _, _, _, err := v.writeNeedle2(newRandomNeedle(4), true, false, false); err != nil { + t.Fatalf("write after the refused commit: %v", err) + } +} + +func TestCommitCompactMakeupDiffFailureKeepsVolumeLoaded(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) + } + defer v.Close() + written := map[uint64]*needle.Needle{} + for i := uint64(1); i <= 3; i++ { + written[i] = newRandomNeedle(i) + if _, _, _, err := v.writeNeedle2(written[i], true, false, false); err != nil { + t.Fatalf("write %d: %v", i, err) + } + } + // Moves needle 3 in the compacted copy, so reads through its map would miss. + if _, err := v.deleteNeedle2(newEmptyNeedle(1)); err != nil { + t.Fatalf("delete: %v", err) + } + delete(written, 1) + if err := v.CompactByIndex(nil); err != nil { + t.Fatalf("compact: %v", err) + } + written[4] = newRandomNeedle(4) + if _, _, _, err := v.writeNeedle2(written[4], true, false, false); err != nil { + t.Fatalf("write during compaction: %v", err) + } + v.lastCompactRevision++ + + if err := v.CommitCompact(); err == nil { + t.Fatal("commit succeeded with a mismatched compaction revision") + } + if _, _, _, _, _, _, ok := v.collectStatus(); !ok { + t.Fatal("volume dropped out of the heartbeat after a failed makeupDiff") + } + for i, want := range written { + got := &needle.Needle{Id: types.NeedleId(i)} + if _, err := v.readNeedle(got, nil, nil); err != nil { + t.Fatalf("read %d after the failed commit: %v", i, err) + } + if !bytes.Equal(got.Data, want.Data) { + t.Fatalf("needle %d data changed after the failed commit", i) + } + } + if _, _, _, err := v.writeNeedle2(newRandomNeedle(5), true, false, false); err != nil { + t.Fatalf("write after the failed commit: %v", err) + } +} + func TestCompactVolumeFilesOffline(t *testing.T) { dir := t.TempDir() location := NewDiskLocation(dir, 10, util.MinFreeSpace{}, dir, "", nil, stats.DefaultDiskIOProbeConfig()) @@ -275,6 +405,9 @@ func TestCleanupCompactRemovesTempFiles(t *testing.T) { if err := tempVolume.cleanupCompact(); err != nil { t.Fatalf("cleanupCompact: %v", err) } + if tempVolume.isCompactionInProgress.Load() { + t.Fatal("cleanupCompact kept the volume claimed") + } if _, err := os.Stat(filepath.Join(dir, "1.cpd")); !os.IsNotExist(err) { t.Fatalf("expected cleanup to remove .cpd, got err=%v", err) diff --git a/weed/storage/volume_write.go b/weed/storage/volume_write.go index bd3bc39c4f..e9c2bada19 100644 --- a/weed/storage/volume_write.go +++ b/weed/storage/volume_write.go @@ -63,7 +63,7 @@ func (v *Volume) destroyLocked(onlyEmpty bool, onlyGarbage bool, keepRemoteData return } if !v.isCompactionInProgress.CompareAndSwap(false, true) { - err = fmt.Errorf("volume %d is compacting", v.Id) + err = v.compactingError() return } storageName, storageKey := v.RemoteStorageNameKey()