shell: delete an empty volume only when every copy is empty

volume.deleteEmpty judged each copy on its own, so it removed an empty
copy while another copy still held data or was offline, leaving that
one as the volume's only copy. Low-disk copies now report
readOnlyCanDelete and passed the check as well. Apply the master
sweep's rule: every copy is listed and every copy is empty.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

weed/shell/command_volume_delete_empty.go

@@ -5,6 +5,8 @@ import (
 	"flag"
 	"io"
 	"log"
+	"maps"
+	"slices"
 	"time"
 
 	"github.com/seaweedfs/seaweedfs/weed/pb"
@@ -31,10 +33,12 @@ func (c *commandVolumeDeleteEmpty) Help() string {
 	volume.deleteEmpty -quietFor=24h -apply
 	volume.deleteEmpty -collectionPattern=important* -quietFor=24h -apply
 
-	This command deletes all empty volumes from one volume server.
+	This command deletes all empty volumes from all volume servers.
 	A volume with no live needles left is empty too, even when its
 	.dat file is still large: compacting it first would only rewrite
 	bytes that are all deleted already.
+	A replicated volume is deleted only when all of its copies are
+	listed and every copy is empty.
 
 `
 }
@@ -76,31 +80,46 @@ func (c *commandVolumeDeleteEmpty) Do(args []string, commandEnv *CommandEnv, wri
 	quietSeconds := int64(*quietPeriod / time.Second)
 	nowUnixSeconds := time.Now().Unix()
 
-	eachDataNode(topologyInfo, func(dc DataCenterId, rack RackId, dn *master_pb.DataNodeInfo) {
-		for _, diskInfo := range dn.DiskInfos {
-			for _, v := range diskInfo.VolumeInfos {
-				if isEmptyVolumeDeleteCandidate(v, quietSeconds, nowUnixSeconds, collectionMatcher) {
-					if *applyBalancing {
-						log.Printf("deleting empty volume %d from %s", v.Id, dn.Id)
-						onlyGarbage := v.FileCount > 0 && v.FileCount <= v.DeleteCount
-						// onlyEmpty stays set so a pre-upgrade server checks
-						// emptiness and refuses instead of deleting unseen.
-						if deleteErr := deleteVolume(context.Background(), commandEnv.option.GrpcDialOption, needle.VolumeId(v.Id),
-							pb.NewServerAddressFromDataNode(dn), true, onlyGarbage, false); deleteErr != nil {
-							err = deleteErr
-						}
-						continue
-					} else {
-						log.Printf("empty volume %d from %s", v.Id, dn.Id)
-					}
-				}
+	volumeReplicas, _ := collectVolumeReplicaLocations(topologyInfo)
+	for _, vid := range slices.Sorted(maps.Keys(volumeReplicas)) {
+		replicas := volumeReplicas[vid]
+		if !isEmptyVolumeCopiesDeleteCandidate(replicas, quietSeconds, nowUnixSeconds, collectionMatcher) {
+			continue
+		}
+		for _, r := range replicas {
+			dn := r.location.dataNode
+			if !*applyBalancing {
+				log.Printf("empty volume %d from %s", vid, dn.Id)
+				continue
+			}
+			log.Printf("deleting empty volume %d from %s", vid, dn.Id)
+			onlyGarbage := r.info.FileCount > 0 && r.info.FileCount <= r.info.DeleteCount
+			// onlyEmpty stays set so a pre-upgrade server checks
+			// emptiness and refuses instead of deleting unseen.
+			if deleteErr := deleteVolume(context.Background(), commandEnv.option.GrpcDialOption, needle.VolumeId(vid),
+				pb.NewServerAddressFromDataNode(dn), true, onlyGarbage, false); deleteErr != nil {
+				err = deleteErr
 			}
 		}
-	})
+	}
 
 	return
 }
 
+// isEmptyVolumeCopiesDeleteCandidate requires every copy to be listed and
+// empty: deleting only the empty ones would leave a copy that holds data, or
+// one that is offline, as the volume's only copy.
+func isEmptyVolumeCopiesDeleteCandidate(replicas []*VolumeReplica, quietSeconds, nowUnixSeconds int64, collectionMatcher *wildcard.CollectionMatcher) bool {
+	for _, r := range replicas {
+		rp, err := super_block.NewReplicaPlacementFromByte(byte(r.info.ReplicaPlacement))
+		if err != nil || len(replicas) < rp.GetCopyCount() ||
+			!isEmptyVolumeDeleteCandidate(r.info, quietSeconds, nowUnixSeconds, collectionMatcher) {
+			return false
+		}
+	}
+	return len(replicas) > 0
+}
+
 func isEmptyVolumeDeleteCandidate(v *master_pb.VolumeInformationMessage, quietSeconds, nowUnixSeconds int64, collectionMatcher *wildcard.CollectionMatcher) bool {
 	return collectionMatcher.Matches(v.Collection) &&
 		// A remote-backed replica shares its cloud object with the others, so

weed/shell/command_volume_delete_empty_test.go

@@ -39,3 +39,46 @@ func TestIsEmptyVolumeDeleteCandidate(t *testing.T) {
 		})
 	}
 }
+
+func TestIsEmptyVolumeCopiesDeleteCandidate(t *testing.T) {
+	matcher, err := wildcard.CompileCollectionMatcher("")
+	assert.NoError(t, err)
+
+	const quietSeconds = int64(60)
+	const now = int64(1_000_000)
+	old := now - quietSeconds - 1
+	copyOf := func(rp uint32, v *master_pb.VolumeInformationMessage) *VolumeReplica {
+		v.ReplicaPlacement = rp
+		return &VolumeReplica{location: &location{dataNode: &master_pb.DataNodeInfo{Id: "dn"}}, info: v}
+	}
+	empty := func() *master_pb.VolumeInformationMessage {
+		return &master_pb.VolumeInformationMessage{Size: super_block.SuperBlockSize, ModifiedAtSecond: old}
+	}
+	lowDisk := func() *master_pb.VolumeInformationMessage {
+		return &master_pb.VolumeInformationMessage{Size: super_block.SuperBlockSize, ReadOnly: true, ReadOnlyCanDelete: true, ModifiedAtSecond: old}
+	}
+	withData := func() *master_pb.VolumeInformationMessage {
+		return &master_pb.VolumeInformationMessage{Size: 1 << 20, FileCount: 3, ModifiedAtSecond: old}
+	}
+	protected := func() *master_pb.VolumeInformationMessage {
+		return &master_pb.VolumeInformationMessage{Size: super_block.SuperBlockSize, ReadOnly: true, ModifiedAtSecond: old}
+	}
+
+	tests := []struct {
+		name     string
+		replicas []*VolumeReplica
+		want     bool
+	}{
+		{"single copy volume", []*VolumeReplica{copyOf(0, empty())}, true},
+		{"every copy empty", []*VolumeReplica{copyOf(1, empty()), copyOf(1, lowDisk())}, true},
+		{"other copy holds data", []*VolumeReplica{copyOf(1, lowDisk()), copyOf(1, withData())}, false},
+		{"other copy is missing", []*VolumeReplica{copyOf(1, lowDisk())}, false},
+		{"other copy is protected read-only", []*VolumeReplica{copyOf(1, empty()), copyOf(1, protected())}, false},
+		{"no copies", nil, false},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			assert.Equal(t, tc.want, isEmptyVolumeCopiesDeleteCandidate(tc.replicas, quietSeconds, now, matcher))
+		})
+	}
+}