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()