worker: project the moved volume when gating on disk fullness (#10171)

The disk-fullness gate only rejected destinations already at/above the mark, so a
server just under it could take a large volume and overshoot. Project the selected
volume's bytes onto the candidate: if the move would cross the mark, drop that
destination for the rest of the cycle and re-pick instead of overshooting. Also
note the per-location capacity-summing assumption on the Rust heartbeat side, to
match the Go store.go comment.

seaweed-volume/src/server/heartbeat.rs

@@ -831,6 +831,9 @@ fn build_heartbeat_with_ec_status(
         }
         *max_volume_counts.entry(disk_type_str.clone()).or_insert(0) += effective_max_count as u32;
         disk_max_by_id[disk_id] = effective_max_count;
+        // Sum capacity per disk type; assumes one location per filesystem. Locations
+        // sharing a mount over-report absolute bytes but not the used ratio the gate
+        // uses. Mirrors weed/storage/store.go.
         *disk_total_bytes.entry(disk_type_str.clone()).or_insert(0) +=
             loc.disk_total_bytes.load(Ordering::Relaxed);
         *disk_free_bytes.entry(disk_type_str).or_insert(0) +=

weed/worker/tasks/balance/detection.go

@@ -205,19 +205,20 @@ func detectForDiskType(diskType string, diskMetrics []*types.VolumeHealthMetrics
 	// planned instead of only its heartbeat-time free space (the shell equivalent
 	// is adjustAfterMove decrementing DiskFreeBytes).
 	plannedBytes := make(map[string]uint64)
+	// destinationFull excludes servers a projected move would push over, for the cycle.
+	destinationFull := make(map[string]bool)
 
-	// destinationDiskTooFull reports whether a server's physical disk is already
-	// at/above the high-water mark, making it ineligible as a move destination.
-	// Judged per server against its own disk, discounting data planned onto it so
-	// far this cycle; servers not reporting disk bytes are never gated (slot-only).
-	destinationDiskTooFull := func(server string) bool {
+	// destinationDiskTooFull reports whether landing incomingBytes on server would put
+	// its disk at/over the mark, net of bytes already planned this cycle; unreported
+	// disks are never gated.
+	destinationDiskTooFull := func(server string, incomingBytes uint64) bool {
 		free := serverDiskFreeBytes[server]
 		if planned := plannedBytes[server]; planned < free {
 			free -= planned
 		} else {
 			free = 0
 		}
-		return balancer.DiskTooFullAfter(serverDiskTotalBytes[server], free, 0, balancer.DefaultMaxDiskUsagePercent)
+		return balancer.DiskTooFullAfter(serverDiskTotalBytes[server], free, incomingBytes, balancer.DefaultMaxDiskUsagePercent)
 	}
 
 	for len(results) < maxResults {
@@ -245,7 +246,7 @@ func detectForDiskType(diskType string, diskMetrics []*types.VolumeHealthMetrics
 			// Min is the emptiest server that can actually receive a volume, so a
 			// physically full server (whose over-set maxVolumeCount makes its slot
 			// utilization look low) is never chosen as the destination.
-			if !destinationDiskTooFull(server) && util < minUtilization {
+			if !destinationFull[server] && !destinationDiskTooFull(server, 0) && util < minUtilization {
 				minUtilization = util
 				minServer = server
 			}
@@ -373,6 +374,16 @@ func detectForDiskType(diskType string, diskMetrics []*types.VolumeHealthMetrics
 			continue
 		}
 
+		// Skip a destination this specific volume would push over the mark, and
+		// re-pick, rather than overshoot.
+		if destinationDiskTooFull(minServer, uint64(selectedVolume.Size)) {
+			glog.V(1).Infof("BALANCE [%s]: skip destination %s: volume %d (%d bytes) would cross %d%% disk usage",
+				diskType, minServer, selectedVolume.VolumeID, selectedVolume.Size, balancer.DefaultMaxDiskUsagePercent)
+			destinationFull[minServer] = true
+			serverCursors[maxServer]-- // retry this volume against another destination
+			continue
+		}
+
 		// Create task targeting minServer — the greedy algorithm's natural choice.
 		// Using minServer instead of letting planBalanceDestination independently
 		// pick a destination ensures that the detection loop's effective counts
@@ -385,7 +396,7 @@ func detectForDiskType(diskType string, diskMetrics []*types.VolumeHealthMetrics
 		// sources are unaffected, so a full server can still be drained.
 		eligibleTargets := make(map[string]int, len(serverVolumeCounts))
 		for s, c := range serverVolumeCounts {
-			if !destinationDiskTooFull(s) {
+			if !destinationFull[s] && !destinationDiskTooFull(s, 0) {
 				eligibleTargets[s] = c
 			}
 		}

weed/worker/tasks/balance/detection_test.go

@@ -465,6 +465,33 @@ func TestDetection_NoDestinationWhenAllDisksFull(t *testing.T) {
 	}
 }
 
+// near-full is at 85% (under the mark) but moving one 100-byte volume onto its
+// 1000-byte disk would reach 95%, so it must not be targeted.
+func TestDetection_ProjectsVolumeOntoNearFullDestination(t *testing.T) {
+	servers := []serverSpec{
+		{id: "src", diskType: "hdd", dc: "dc1", rack: "rack1", maxVolumes: 10},
+		{id: "near-full", diskType: "hdd", dc: "dc1", rack: "rack1", maxVolumes: 1000, diskTotalBytes: 1000, diskFreeBytes: 150},
+	}
+	metrics := make([]*types.VolumeHealthMetrics, 8)
+	for i := range metrics {
+		metrics[i] = &types.VolumeHealthMetrics{
+			VolumeID: uint32(i + 1), Server: "src", ServerAddress: "src:8080",
+			DiskType: "hdd", Collection: "c1", Size: 100, DataCenter: "dc1", Rack: "rack1",
+		}
+	}
+
+	at := buildTopology(servers, metrics)
+	clusterInfo := &types.ClusterInfo{ActiveTopology: at}
+
+	tasks, _, err := Detection(metrics, clusterInfo, defaultConf(), 100)
+	if err != nil {
+		t.Fatalf("Detection failed: %v", err)
+	}
+	if len(tasks) != 0 {
+		t.Fatalf("expected no tasks: the projected move would cross 90%% disk usage, got %d", len(tasks))
+	}
+}
+
 func TestDetection_SkipsRemoteTieredVolumes(t *testing.T) {
 	metrics := []*types.VolumeHealthMetrics{}