use atomic.Bool for compaction flags to prevent concurrent vacuum races

The isCompacting and isCommitCompacting flags were plain bools
read and written from multiple goroutines without synchronization.
This allowed concurrent vacuums on the same volume to pass the
guard checks and run simultaneously, leading to the nil pointer
crash. Using atomic.Bool with CompareAndSwap ensures only one
compaction or commit can run per volume at a time.

Fixes #8591

weed/storage/disk_location.go

@@ -383,7 +383,7 @@ func (l *DiskLocation) UnloadVolume(vid needle.VolumeId) error {
 func (l *DiskLocation) unmountVolumeByCollection(collectionName string) map[needle.VolumeId]*Volume {
 	deltaVols := make(map[needle.VolumeId]*Volume, 0)
 	for k, v := range l.volumes {
-		if v.Collection == collectionName && !v.isCompacting && !v.isCommitCompacting {
+		if v.Collection == collectionName && !v.isCompacting.Load() && !v.isCommitCompacting.Load() {
 			deltaVols[k] = v
 		}
 	}

weed/storage/volume.go

@@ -5,6 +5,7 @@ import (
 	"path"
 	"strconv"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
@@ -45,8 +46,8 @@ type Volume struct {
 	lastCompactRevision    uint16
 	ldbTimeout             int64
 
-	isCompacting       bool
-	isCommitCompacting bool
+	isCompacting       atomic.Bool
+	isCommitCompacting atomic.Bool
 
 	volumeInfoRWLock sync.RWMutex
 	volumeInfo       *volume_server_pb.VolumeInfo
@@ -231,7 +232,7 @@ func (v *Volume) Close() {
 }
 
 func (v *Volume) doClose() {
-	for v.isCommitCompacting {
+	for v.isCommitCompacting.Load() {
 		time.Sleep(521 * time.Millisecond)
 		glog.Warningf("Volume Close wait for compaction %d", v.Id)
 	}

weed/storage/volume_vacuum.go

@@ -64,14 +64,11 @@ func (v *Volume) CompactByVolumeData(opts *CompactOptions) error {
 	//v.accessLock.Lock()
 	//defer v.accessLock.Unlock()
 	//glog.V(3).Infof("Got Compaction lock...")
-	if v.isCompacting || v.isCommitCompacting {
+	if v.isCommitCompacting.Load() || !v.isCompacting.CompareAndSwap(false, true) {
 		glog.V(0).Infof("Volume %d is already compacting...", v.Id)
 		return nil
 	}
-	v.isCompacting = true
-	defer func() {
-		v.isCompacting = false
-	}()
+	defer v.isCompacting.Store(false)
 
 	v.lastCompactIndexOffset = v.IndexFileSize()
 	v.lastCompactRevision = v.SuperBlock.CompactionRevision
@@ -102,14 +99,11 @@ func (v *Volume) CompactByIndex(opts *CompactOptions) error {
 	}
 	glog.V(3).Infof("Compact2 volume %d ...", v.Id)
 
-	if v.isCompacting || v.isCommitCompacting {
+	if v.isCommitCompacting.Load() || !v.isCompacting.CompareAndSwap(false, true) {
 		glog.V(0).Infof("Volume %d is already compacting2 ...", v.Id)
 		return nil
 	}
-	v.isCompacting = true
-	defer func() {
-		v.isCompacting = false
-	}()
+	defer v.isCompacting.Store(false)
 
 	v.lastCompactIndexOffset = v.IndexFileSize()
 	v.lastCompactRevision = v.SuperBlock.CompactionRevision
@@ -142,14 +136,11 @@ func (v *Volume) CommitCompact() error {
 	}
 	glog.V(0).Infof("Committing volume %d vacuuming...", v.Id)
 
-	if v.isCommitCompacting {
+	if !v.isCommitCompacting.CompareAndSwap(false, true) {
 		glog.V(0).Infof("Volume %d is already commit compacting ...", v.Id)
 		return nil
 	}
-	v.isCommitCompacting = true
-	defer func() {
-		v.isCommitCompacting = false
-	}()
+	defer v.isCommitCompacting.Store(false)
 
 	v.dataFileAccessLock.Lock()
 	defer v.dataFileAccessLock.Unlock()

weed/storage/volume_write.go

@@ -70,7 +70,7 @@ func (v *Volume) Destroy(onlyEmpty bool) (err error) {
 			return
 		}
 	}
-	if v.isCompacting || v.isCommitCompacting {
+	if v.isCompacting.Load() || v.isCommitCompacting.Load() {
 		err = fmt.Errorf("volume %d is compacting", v.Id)
 		return
 	}