topology: clamp the deleted-vs-total subtractions in volume stats (#10633)

VolumeLocationList.Stats subtracts the deleted figures from the totals to
report live size and needle count. Both deleted figures are maintained as
counters independent of the totals they come off, so either can transiently
exceed its total, and neither subtraction was clamped.

Unclamped, the size wraps to ~16 EB. The count is signed so it merely goes
negative, but VolumeLayout.Stats converts it with uint64(fileCount), which
turns it into ~1.8e19 just the same. Either one swamps the cluster totals
behind /dir/status, /vol/status and Topology.CollectionVolumeStats.

commandFsMergeVolumes.getVolumeSize had the same unclamped subtraction, where
a wrapped size reads as a volume far too large to join any merge plan.

Clamped to zero, matching the guards already in CollectionInfo.LogicalSize
and the admin server's logical-size accumulator.

weed/shell/command_fs_merge_volumes.go

@@ -465,7 +465,12 @@ func (c *commandFsMergeVolumes) getVolumeSizeBasedOnPlan(plan map[needle.VolumeI
 	return size
 }
 
+// getVolumeSize is the volume's live data size, clamped since
+// DeletedByteCount can transiently exceed Size.
 func (c *commandFsMergeVolumes) getVolumeSize(volume *master_pb.VolumeInformationMessage) uint64 {
+	if volume.Size < volume.DeletedByteCount {
+		return 0
+	}
 	return volume.Size - volume.DeletedByteCount
 }
 

weed/shell/command_fs_merge_volumes_test.go

@@ -123,3 +123,18 @@ func TestCreateMergePlan_DirectedRejectsIneligible(t *testing.T) {
 		})
 	}
 }
+
+// A volume reporting more deleted bytes than it holds must read as empty.
+func TestGetVolumeSize_ClampsDeletedOverSize(t *testing.T) {
+	c := newMergeCmd(250000)
+
+	overDeleted := &master_pb.VolumeInformationMessage{Id: 1, Size: 1000, DeletedByteCount: 4000}
+	if got := c.getVolumeSize(overDeleted); got != 0 {
+		t.Errorf("expected 0 for a volume with more deleted than stored, got %d", got)
+	}
+
+	healthy := &master_pb.VolumeInformationMessage{Id: 2, Size: 3000, DeletedByteCount: 1000}
+	if got := c.getVolumeSize(healthy); got != 2000 {
+		t.Errorf("expected 2000, got %d", got)
+	}
+}

weed/topology/topology_stats_test.go

@@ -111,3 +111,25 @@ func TestCollectionVolumeStatsWithEcVolumes(t *testing.T) {
 		t.Errorf("ec stats query should not create a phantom collection")
 	}
 }
+
+func TestCollectionVolumeStatsClampsDeletedOverTotals(t *testing.T) {
+	topo := NewTopology("weedfs", sequence.NewMemorySequencer(), 32*1024, 5, false)
+
+	rack := topo.GetOrCreateDataCenter("dc1").GetOrCreateRack("rack1")
+	dn := rack.GetOrCreateDataNode("127.0.0.1", 34534, 0, "127.0.0.1", "", map[string]uint32{"": 25})
+
+	// Volume 1 reports more deleted than it holds, on both counters.
+	topo.SyncDataNodeRegistration([]*master_pb.VolumeInformationMessage{
+		{Id: 1, Size: 1000, FileCount: 10, DeletedByteCount: 4000, DeleteCount: 40, Version: uint32(needle.GetCurrentVersion())},
+		{Id: 2, Size: 3000, FileCount: 30, DeletedByteCount: 1000, DeleteCount: 10, Version: uint32(needle.GetCurrentVersion())},
+	}, dn)
+
+	// VolumeLocationList.Stats only counts nodes connected for over a minute
+	dn.LastSeen = time.Now().Unix() - 61
+
+	stats := topo.CollectionVolumeStats("")
+	// Volume 1 contributes nothing rather than wrapping.
+	assert(t, "used size", int(stats.UsedSize), 2000)
+	assert(t, "logical used size", int(stats.LogicalUsedSize), 2000)
+	assert(t, "file count", int(stats.FileCount), 20)
+}

weed/topology/volume_location_list.go

@@ -96,13 +96,21 @@ func (dnll *VolumeLocationList) Refresh(freshThreshHold int64) {
 	}
 }
 
-// Stats returns logic size and count
+// Stats returns logic size and count. Both subtractions are clamped: the
+// deleted counters are maintained apart from the totals they come off and can
+// transiently exceed them.
 func (dnll *VolumeLocationList) Stats(vid needle.VolumeId, freshThreshHold int64) (size uint64, fileCount int) {
 	for _, dnl := range dnll.list {
 		if dnl.LastSeen < freshThreshHold {
 			vinfo, err := dnl.GetVolumesById(vid)
 			if err == nil {
-				return (vinfo.Size - vinfo.DeletedByteCount), vinfo.FileCount - vinfo.DeleteCount
+				if vinfo.Size > vinfo.DeletedByteCount {
+					size = vinfo.Size - vinfo.DeletedByteCount
+				}
+				if vinfo.FileCount > vinfo.DeleteCount {
+					fileCount = vinfo.FileCount - vinfo.DeleteCount
+				}
+				return size, fileCount
 			}
 		}
 	}