ec_balance: drain a server holding far more shards than its siblings (#12477)

* ecbalancer: drain a full server through any target that can take a shard

The global phase paired the fullest server with the single emptiest one and
stopped for the whole rack when that pair had no legal move. Where every
volume already has a shard on each machine, no cross-machine move keeps the
spread, so on hosts running one volume server per disk the first server's
disk kept its pile while its siblings on the same machine stayed nearly
empty: the phase never planned a move.

Try sources fullest first and targets emptiest first and take the first pair
with a legal move, so such a server drains onto its own machine's emptier
servers. A same-machine move goes only to a server holding none of the
volume, so one disk never carries two of its shards. Per-machine counts are
indexed once per rack and pairs without a move are remembered until a move
can change them, so an uncapped drain stays fast.

Claude-Session: https://claude.ai/code/session_011UpvafcfUuC2mLxNAXLQJi

* ec_balance: keep global moves ahead of within-rack moves under the detection cap

Detection truncated the plan to the per-cycle job limit in planner order,
which puts the global moves last. A long within-rack backlog pushed all of
them past the limit, so even a planned global move never ran. Order dedup
and cross-rack first, then the capped global phase, then within-rack.

Claude-Session: https://claude.ai/code/session_011UpvafcfUuC2mLxNAXLQJi

* ec_balance: keep each volume's moves in plan order when promoting global moves

The admin scheduler runs only the first proposal per volume each cycle. A
global move promoted ahead of its volume's within-rack moves can take a shard
that one of those moves placed, so it ran while the shard was still on the
original server and failed.

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

* ecbalancer: order global targets by fullness after taking the shard

The search stops at the first target that would end up fuller than the
source, which only holds when targets are ordered by that same value. With
unequal capacities an empty one-slot server sorted first and ended the search
before a larger server that could take the shard.

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

* ecbalancer: free the source disk slot on a global move

The global phase cleared the moved shard's bits but left the source disk's
free slots and shard count unchanged, unlike every other phase. Use
releaseShard so a later pick of that server sees the freed disk.

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

* ec_balance: test that a capped detection keeps a global move

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

* ecbalancer: test the same-machine move skips a sibling holding the volume

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

---------

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

weed/storage/erasure_coding/ecbalancer/balancer.go

@@ -937,115 +937,190 @@ func detectGlobalImbalance(nodes map[string]*Node, racks map[string]*rack, diskT
 		if iterations <= 0 {
 			iterations = totalShards
 		}
+		spread := newRackSpread(r)
+		blocked := make(blockedPairs)
 		for i := 0; i < iterations; i++ {
-			var minNode, maxNode *Node
-			minUtil := math.Inf(1)
-			maxUtil := -1.0
-			var minCount, maxCount int
-			for _, nodeID := range sortedNodeKeys(r.nodes) {
-				count := nodeShardCounts[nodeID]
-				node := r.nodes[nodeID]
-				capacity := nodeCapacity[nodeID]
-				if capacity <= 0 {
-					continue
-				}
-				util := float64(count) / float64(capacity)
-				if util < minUtil && node.receiveSlots() > 0 {
-					minUtil, minCount, minNode = util, count, node
-				}
-				if util > maxUtil {
-					maxUtil, maxCount, maxNode = util, count, node
-				}
-			}
-			if maxNode == nil || minNode == nil || maxNode.id == minNode.id {
-				break
-			}
-
-			maxCap := nodeCapacity[maxNode.id]
-			minCap := nodeCapacity[minNode.id]
-			if maxCap <= 0 || minCap <= 0 {
-				break
-			}
-			if float64(minCount+1)/float64(minCap) > float64(maxCount-1)/float64(maxCap) {
-				break
-			}
-
-			// Prefer a volume absent from the destination's machine (pass 0) before
-			// adding to one already there (pass 1), to keep volumes spread.
-			moved := false
-			for pass := 0; pass < 2 && !moved; pass++ {
-				for _, vk := range sortedVolumeKeys(maxNode.shards) {
-					if moved {
-						break
-					}
-					info := maxNode.shards[vk]
-					minInfo := minNode.shards[vk]
-					volumeOnMinMachine := machineHoldsVolume(r, minNode.host, vk)
-					if pass == 0 && volumeOnMinMachine {
-						continue // pass 0: only volumes absent from the destination machine
-					}
-					if pass == 1 {
-						if !volumeOnMinMachine {
-							continue
-						}
-						// Protect the volume's machine spread only where it's achievable
-						// (enough machines for this rack's shards to each stay within
-						// parity); there a cross-machine load move is allowed only if it
-						// doesn't raise the destination machine's count past the source's.
-						// Where it isn't achievable, capacity rules and any leveling move
-						// is fine. Feasibility uses the rack's shards, not the whole volume.
-						parity := parityShardsByVolume[vk]
-						spreadFeasible := parity > 0 && rackMachineCount >= ceilDivide(rackVolumeShardCount(r, vk), parity)
-						if spreadFeasible && minNode.host != maxNode.host &&
-							machineVolumeCount(r, minNode.host, vk) >= machineVolumeCount(r, maxNode.host, vk) {
-							continue
-						}
-					}
-					// Walk the volume's actual shard bitmap so custom ratios with more
-					// than the standard total (ids 14..MaxShardCount-1) are candidates too.
-					for sid := range info.shardBits.All() {
-						shardID := int(sid)
-						if minInfo != nil && minInfo.shardBits.Has(sid) {
-							continue
-						}
-						dataShards := dataShardsByVolume[vk]
-						if dataShards <= 0 {
-							dataShards = erasure_coding.DataShardsCount
-						}
-						destDisk := pickBestDiskOnNode(minNode, vk, diskType, shardID, dataShards)
-						moves = append(moves, &move{
-							volumeID:   vk.vid,
-							shardID:    shardID,
-							collection: vk.collection,
-							source:     maxNode,
-							sourceDisk: shardDiskID(maxNode, vk, shardID),
-							target:     minNode,
-							targetDisk: destDisk,
-							phase:      "global",
-						})
-						info.shardBits = info.shardBits.Clear(sid)
-						for diskID := range info.diskShardBits {
-							info.diskShardBits[diskID] = info.diskShardBits[diskID].Clear(sid)
-						}
-						reserveShard(minNode, vk, shardID, destDisk)
-						nodeShardCounts[maxNode.id]--
-						nodeShardCounts[minNode.id]++
-						maxNode.freeSlots++
-						minNode.freeSlots--
-						moved = true
-						break
-					}
-				}
-			}
-			if !moved {
+			m := planGlobalMove(r, spread, blocked, rackMachineCount, nodeShardCounts, nodeCapacity, diskType, dataShardsByVolume, parityShardsByVolume)
+			if m == nil {
 				break
 			}
+			moves = append(moves, m)
+			vk := volKey{collection: m.collection, vid: m.volumeID}
+			releaseShard(m.source, vk, m.shardID)
+			reserveShard(m.target, vk, m.shardID, m.targetDisk)
+			spread.move(vk, m.source.host, m.target.host)
+			blocked.forget(m.source, m.target)
+			nodeShardCounts[m.source.id]--
+			nodeShardCounts[m.target.id]++
+			m.source.freeSlots++
+			m.target.freeSlots--
 		}
 	}
 
 	return moves
 }
 
+// rackSpread indexes each volume's shard counts in a rack, per machine and in
+// total, so the global phase can check spread without rescanning the rack.
+type rackSpread struct {
+	machine map[volKey]map[string]int
+	rack    map[volKey]int
+}
+
+func newRackSpread(r *rack) rackSpread {
+	s := rackSpread{machine: make(map[volKey]map[string]int), rack: make(map[volKey]int)}
+	for _, n := range r.nodes {
+		for vk, info := range n.shards {
+			c := info.shardBits.Count()
+			if c == 0 {
+				continue
+			}
+			if s.machine[vk] == nil {
+				s.machine[vk] = make(map[string]int)
+			}
+			s.machine[vk][n.host] += c
+			s.rack[vk] += c
+		}
+	}
+	return s
+}
+
+// move records one shard of vk leaving fromHost for toHost within the rack.
+func (s rackSpread) move(vk volKey, fromHost, toHost string) {
+	s.machine[vk][fromHost]--
+	s.machine[vk][toHost]++
+}
+
+// blockedPairs remembers source/target pairs with no movable shard, so a long
+// drain does not rescan them every iteration.
+type blockedPairs map[[2]*Node]bool
+
+// forget drops the pairs a move from source to target can unblock. Losing a
+// shard only narrows source's options and gaining one only narrows target's, so
+// within a machine only target as a source (it holds a new shard) and source as
+// a target (it no longer holds one) can open up. A move between machines also
+// changes the volume's count on both machines, which every pair touching either
+// machine checks.
+func (b blockedPairs) forget(source, target *Node) {
+	crossMachine := source.host != target.host
+	for pair := range b {
+		from, to := pair[0], pair[1]
+		if from == target || to == source ||
+			(crossMachine && (from.host == source.host || from.host == target.host || to.host == source.host || to.host == target.host)) {
+			delete(b, pair)
+		}
+	}
+}
+
+// planGlobalMove picks one shard to move from a fuller node to an emptier one in
+// the rack, or returns nil when no move would narrow the gap. Sources are tried
+// fullest first and destinations emptiest first once they hold the shard, so a
+// node whose shards cannot go to the emptiest node still drains to the next one
+// that can take them, such as another disk on its own machine.
+func planGlobalMove(r *rack, spread rackSpread, blocked blockedPairs, rackMachineCount int, counts, capacity map[string]int, diskType string, dataShardsByVolume, parityShardsByVolume map[volKey]int) *move {
+	util := func(n *Node, delta int) float64 {
+		return float64(counts[n.id]+delta) / float64(capacity[n.id])
+	}
+	var sources, targets []*Node
+	for _, n := range sortedNodeSlice(r.nodes) {
+		if capacity[n.id] <= 0 {
+			continue
+		}
+		sources = append(sources, n)
+		if n.receiveSlots() > 0 {
+			targets = append(targets, n)
+		}
+	}
+	sort.SliceStable(sources, func(i, j int) bool { return util(sources[i], 0) > util(sources[j], 0) })
+	sort.SliceStable(targets, func(i, j int) bool { return util(targets[i], 1) < util(targets[j], 1) })
+	for _, source := range sources {
+		for _, target := range targets {
+			if target.id == source.id {
+				continue
+			}
+			// Targets are ordered by their fullness after taking the shard, so once
+			// one would end up fuller than the source, every later one would too.
+			if util(target, 1) > util(source, -1) {
+				break
+			}
+			pair := [2]*Node{source, target}
+			if blocked[pair] {
+				continue
+			}
+			if m := pickGlobalShard(spread, rackMachineCount, source, target, diskType, dataShardsByVolume, parityShardsByVolume); m != nil {
+				return m
+			}
+			blocked[pair] = true
+		}
+	}
+	return nil
+}
+
+// pickGlobalShard returns a move of one of source's shards to target that keeps
+// each volume spread across machines, or nil.
+func pickGlobalShard(spread rackSpread, rackMachineCount int, source, target *Node, diskType string, dataShardsByVolume, parityShardsByVolume map[volKey]int) *move {
+	sameMachine := source.host == target.host
+	// Prefer a volume absent from the target's machine (pass 0) before adding to
+	// one already there (pass 1), to keep volumes spread.
+	for pass := 0; pass < 2; pass++ {
+		for _, vk := range sortedVolumeKeys(source.shards) {
+			info := source.shards[vk]
+			targetInfo := target.shards[vk]
+			volumeOnTargetMachine := spread.machine[vk][target.host] > 0
+			if pass == 0 && volumeOnTargetMachine {
+				continue // pass 0: only volumes absent from the target machine
+			}
+			if pass == 1 {
+				if !volumeOnTargetMachine {
+					continue
+				}
+				if sameMachine {
+					// The machine's count is unchanged; keep the volume on as many
+					// of the machine's disks as before by not doubling up on one.
+					if targetInfo != nil && targetInfo.shardBits.Count() > 0 {
+						continue
+					}
+				} else {
+					// Protect the volume's machine spread only where it's achievable
+					// (enough machines for this rack's shards to each stay within
+					// parity); there a cross-machine load move is allowed only if it
+					// doesn't raise the target machine's count past the source's.
+					// Where it isn't achievable, capacity rules and any leveling move
+					// is fine. Feasibility uses the rack's shards, not the whole volume.
+					parity := parityShardsByVolume[vk]
+					spreadFeasible := parity > 0 && rackMachineCount >= ceilDivide(spread.rack[vk], parity)
+					if spreadFeasible && spread.machine[vk][target.host] >= spread.machine[vk][source.host] {
+						continue
+					}
+				}
+			}
+			// Walk the volume's actual shard bitmap so custom ratios with more
+			// than the standard total (ids 14..MaxShardCount-1) are candidates too.
+			for sid := range info.shardBits.All() {
+				if targetInfo != nil && targetInfo.shardBits.Has(sid) {
+					continue
+				}
+				shardID := int(sid)
+				dataShards := dataShardsByVolume[vk]
+				if dataShards <= 0 {
+					dataShards = erasure_coding.DataShardsCount
+				}
+				return &move{
+					volumeID:   vk.vid,
+					shardID:    shardID,
+					collection: vk.collection,
+					source:     source,
+					sourceDisk: shardDiskID(source, vk, shardID),
+					target:     target,
+					targetDisk: pickBestDiskOnNode(target, vk, diskType, shardID, dataShards),
+					phase:      "global",
+				}
+			}
+		}
+	}
+	return nil
+}
+
 // shardsByGroup classifies a volume's shards into data (id < dataShards) and
 // parity buckets, grouped by key(node).
 func shardsByGroup(vk volKey, nodes map[string]*Node, dataShards int, key func(*Node) string) (dataPer, parityPer map[string][]int) {

weed/storage/erasure_coding/ecbalancer/balancer_test.go

@@ -990,3 +990,126 @@ func TestNoReceiveDiskLeavesSiblingsEligible(t *testing.T) {
 		t.Fatal("a node with one refused disk should still receive on its other disks")
 	}
 }
+
+// TestGlobalDrainsToSameMachineWhenOtherMachinesAreBlocked: hosts run one server
+// per disk, and every volume has one shard on each machine, all on the first
+// server. The emptiest server is on another machine, where any shard would
+// concentrate its volume, so the fullest server must drain onto its own
+// machine's empty server instead of the phase giving up for the whole rack.
+func TestGlobalDrainsToSameMachineWhenOtherMachinesAreBlocked(t *testing.T) {
+	topo := NewTopology()
+	mk := func(id, host string, free int) *Node {
+		n := topo.AddNode(id, "dc1", "dc1:rack1", free)
+		n.SetHost(host)
+		return n
+	}
+	a1, b1, c1 := mk("a1", "boxA", 2), mk("b1", "boxB", 4), mk("c1", "boxC", 4)
+	a2, c2 := mk("a2", "boxA", 40), mk("c2", "boxC", 40)
+	mk("b2", "boxB", 40) // the emptiest server, on another machine than a1
+	a2.AddShards(99, "col1", 0, bits(0))
+	c2.AddShards(99, "col1", 0, bits(1))
+	data, parity := map[volKey]int{{"col1", 99}: 2}, map[volKey]int{{"col1", 99}: 1}
+	for vid := uint32(1); vid <= 6; vid++ {
+		a1.AddShards(vid, "col1", 0, bits(0))
+		b1.AddShards(vid, "col1", 0, bits(1))
+		c1.AddShards(vid, "col1", 0, bits(2))
+		data[volKey{"col1", vid}], parity[volKey{"col1", vid}] = 2, 1
+	}
+
+	moves := detectGlobalImbalance(topo.nodes, buildRacks(topo.nodes), "", 0.2, data, parity, 10, true)
+	if len(moves) == 0 {
+		t.Fatal("no global moves: the fullest server never drained")
+	}
+	if moves[0].source != a1 || moves[0].target != a2 {
+		t.Errorf("first move %s -> %s, want a1 -> a2", moves[0].source.id, moves[0].target.id)
+	}
+	held := map[string]map[uint32]int{}
+	for _, m := range moves {
+		if m.source.host != m.target.host {
+			t.Errorf("cross-machine move %d.%d %s -> %s concentrates the volume", m.volumeID, m.shardID, m.source.id, m.target.id)
+		}
+		if held[m.target.id] == nil {
+			held[m.target.id] = map[uint32]int{}
+		}
+		if held[m.target.id][m.volumeID]++; held[m.target.id][m.volumeID] > 1 {
+			t.Errorf("%s received two shards of volume %d", m.target.id, m.volumeID)
+		}
+	}
+}
+
+// TestGlobalTriesTargetsByFullnessAfterTheMove: a one-slot server is the
+// emptiest but would be full after one shard, which must not stop the search
+// before a larger server that can take the shard.
+func TestGlobalTriesTargetsByFullnessAfterTheMove(t *testing.T) {
+	topo := NewTopology()
+	src := topo.AddNode("src", "dc1", "dc1:rack1", 2)
+	topo.AddNode("tiny", "dc1", "dc1:rack1", 1)
+	big := topo.AddNode("big", "dc1", "dc1:rack1", 9)
+	data, parity := map[volKey]int{}, map[volKey]int{}
+	for vid := uint32(1); vid <= 9; vid++ {
+		n := src
+		if vid == 9 {
+			n = big
+		}
+		n.AddShards(vid, "col1", 0, bits(0))
+		data[volKey{"col1", vid}], parity[volKey{"col1", vid}] = 2, 1
+	}
+
+	moves := detectGlobalImbalance(topo.nodes, buildRacks(topo.nodes), "", 0.2, data, parity, 10, true)
+	if len(moves) == 0 {
+		t.Fatal("no global moves: the one-slot server stopped the search")
+	}
+	for _, m := range moves {
+		if m.source != src || m.target != big {
+			t.Errorf("move %d.%d %s -> %s, want src -> big", m.volumeID, m.shardID, m.source.id, m.target.id)
+		}
+	}
+}
+
+// TestGlobalMoveFreesTheSourceDisk: a shard moved off a server frees its disk
+// slot, so a later pick of that server can choose the disk again.
+func TestGlobalMoveFreesTheSourceDisk(t *testing.T) {
+	topo := NewTopology()
+	src := topo.AddNode("src", "dc1", "dc1:rack1", 0)
+	src.AddDisk(1, "", 0, 4)
+	topo.AddNode("dst", "dc1", "dc1:rack1", 10)
+	data, parity := map[volKey]int{}, map[volKey]int{}
+	for vid := uint32(1); vid <= 4; vid++ {
+		src.AddShards(vid, "col1", 1, bits(0))
+		data[volKey{"col1", vid}], parity[volKey{"col1", vid}] = 2, 1
+	}
+
+	moves := detectGlobalImbalance(topo.nodes, buildRacks(topo.nodes), "", 0.2, data, parity, 1, true)
+	if len(moves) != 1 {
+		t.Fatalf("got %d moves, want 1", len(moves))
+	}
+	if d := src.disks[1]; d.freeSlots != 1 || d.shardCount != 3 {
+		t.Errorf("source disk after the move: %d free, %d shards; want 1 free, 3 shards", d.freeSlots, d.shardCount)
+	}
+}
+
+// TestGlobalSameMachineMoveSkipsASiblingHoldingTheVolume: a same-machine move
+// must not put a second shard of a volume on a server that already holds one.
+func TestGlobalSameMachineMoveSkipsASiblingHoldingTheVolume(t *testing.T) {
+	topo := NewTopology()
+	a1 := topo.AddNode("a1", "dc1", "dc1:rack1", 1)
+	a2 := topo.AddNode("a2", "dc1", "dc1:rack1", 10)
+	a1.SetHost("boxA")
+	a2.SetHost("boxA")
+	data, parity := map[volKey]int{}, map[volKey]int{}
+	for vid := uint32(1); vid <= 3; vid++ {
+		a1.AddShards(vid, "col1", 0, bits(0))
+		data[volKey{"col1", vid}], parity[volKey{"col1", vid}] = 2, 1
+	}
+	a2.AddShards(1, "col1", 0, bits(1))
+
+	moves := detectGlobalImbalance(topo.nodes, buildRacks(topo.nodes), "", 0.2, data, parity, 10, true)
+	if len(moves) == 0 {
+		t.Fatal("no global moves")
+	}
+	for _, m := range moves {
+		if m.volumeID == 1 {
+			t.Errorf("moved shard 1.%d %s -> %s onto a server holding the volume", m.shardID, m.source.id, m.target.id)
+		}
+	}
+}

weed/worker/tasks/ec_balance/detection.go

@@ -97,6 +97,12 @@ func Detection(
 		return nil, false, nil
 	}
 
+	// The global phase is capped per rack while the per-volume phases are not,
+	// so order it ahead of the within-rack moves: otherwise a long within-rack
+	// backlog pushes every global move past the detection cap, and a server that
+	// holds far more shards than its peers never drains.
+	sortMovesByPhase(moves)
+
 	hasMore := false
 	if maxResults > 0 && len(moves) > maxResults {
 		moves = moves[:maxResults]
@@ -339,6 +345,48 @@ func countEcShards(ecShardInfos []*master_pb.VolumeEcShardInformationMessage) in
 	return count
 }
 
+// sortMovesByPhase orders a plan for truncation, keeping the planner's order
+// within a phase: dedup and cross-rack first, then the capped global phase,
+// then within-rack. A volume's moves keep their plan order: a global move can
+// take a shard that an earlier within-rack move placed, and the scheduler runs
+// only a volume's first move per cycle.
+func sortMovesByPhase(moves []ecbalancer.Move) {
+	type volumeKey struct {
+		collection string
+		vid        uint32
+	}
+	volumeOrder := make(map[volumeKey]int)
+	order := make([]int, len(moves))
+	for i, m := range moves {
+		vk := volumeKey{m.Collection, m.VolumeID}
+		volumeOrder[vk] = max(volumeOrder[vk], movePhaseOrder(m.Phase))
+		order[i] = volumeOrder[vk]
+	}
+	idx := make([]int, len(moves))
+	for i := range idx {
+		idx[i] = i
+	}
+	slices.SortStableFunc(idx, func(a, b int) int { return order[a] - order[b] })
+	sorted := make([]ecbalancer.Move, len(moves))
+	for i, j := range idx {
+		sorted[i] = moves[j]
+	}
+	copy(moves, sorted)
+}
+
+func movePhaseOrder(phase string) int {
+	switch phase {
+	case "dedup":
+		return 0
+	case "cross_rack":
+		return 1
+	case "global":
+		return 2
+	default:
+		return 3
+	}
+}
+
 func movePhasePriority(phase string) types.TaskPriority {
 	switch phase {
 	case "dedup":

weed/worker/tasks/ec_balance/detection_test.go

@@ -3,8 +3,11 @@ package ec_balance
 import (
 	"context"
 	"net"
+	"slices"
+	"strings"
 	"testing"
 
+	"github.com/seaweedfs/seaweedfs/weed/admin/topology"
 	"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
 	"github.com/seaweedfs/seaweedfs/weed/storage/erasure_coding"
 	"github.com/seaweedfs/seaweedfs/weed/storage/erasure_coding/ecbalancer"
@@ -154,3 +157,99 @@ func TestMovePhasePriority(t *testing.T) {
 
 // keep the erasure_coding import meaningful for future adapter tests
 var _ = erasure_coding.DataShardsCount
+
+// TestSortMovesByPhaseKeepsGlobalAheadOfWithinRack: a long within-rack backlog
+// must not push the capped global moves past the detection cap.
+func TestSortMovesByPhaseKeepsGlobalAheadOfWithinRack(t *testing.T) {
+	moves := []ecbalancer.Move{
+		{VolumeID: 1, Phase: "within_rack"},
+		{VolumeID: 2, Phase: "dedup"},
+		{VolumeID: 3, Phase: "within_rack"},
+		{VolumeID: 4, Phase: "global"},
+		{VolumeID: 5, Phase: "cross_rack"},
+		{VolumeID: 6, Phase: "global"},
+	}
+	sortMovesByPhase(moves)
+	var got []uint32
+	for _, m := range moves {
+		got = append(got, m.VolumeID)
+	}
+	if want := []uint32{2, 5, 4, 6, 1, 3}; !slices.Equal(got, want) {
+		t.Errorf("order = %v, want %v", got, want)
+	}
+}
+
+// TestSortMovesByPhaseKeepsAVolumesMovesInPlanOrder: a global move of a shard
+// that a within-rack move placed must not run ahead of that move.
+func TestSortMovesByPhaseKeepsAVolumesMovesInPlanOrder(t *testing.T) {
+	moves := []ecbalancer.Move{
+		{VolumeID: 1, ShardID: 3, SourceNode: "within 1", Phase: "within_rack"},
+		{VolumeID: 2, SourceNode: "within 2", Phase: "within_rack"},
+		{VolumeID: 1, ShardID: 3, SourceNode: "global 1", Phase: "global"},
+		{VolumeID: 3, SourceNode: "global 3", Phase: "global"},
+		{VolumeID: 2, Collection: "other", SourceNode: "global other 2", Phase: "global"},
+	}
+	sortMovesByPhase(moves)
+	var got []string
+	for _, m := range moves {
+		got = append(got, m.SourceNode)
+	}
+	if want := []string{"global 3", "global other 2", "within 1", "within 2", "global 1"}; !slices.Equal(got, want) {
+		t.Errorf("order = %q, want %q", got, want)
+	}
+}
+
+// TestDetectionKeepsGlobalMovesUnderTheCap: the within-rack moves that come
+// first in plan order must not push every global move past maxResults.
+func TestDetectionKeepsGlobalMovesUnderTheCap(t *testing.T) {
+	disk := func(shards ...*master_pb.VolumeEcShardInformationMessage) map[string]*master_pb.DiskInfo {
+		return map[string]*master_pb.DiskInfo{"": {Type: "", MaxVolumeCount: 100, EcShardInfos: shards}}
+	}
+	// Volume 100 sits on one machine (within-rack moves); volumes 1-20 load one
+	// server (global moves).
+	var loaded []*master_pb.VolumeEcShardInformationMessage
+	for vid := uint32(1); vid <= 20; vid++ {
+		loaded = append(loaded, &master_pb.VolumeEcShardInformationMessage{Id: vid, Collection: "col1", EcIndexBits: 1})
+	}
+	topoInfo := &master_pb.TopologyInfo{
+		DataCenterInfos: []*master_pb.DataCenterInfo{{
+			Id: "dc1",
+			RackInfos: []*master_pb.RackInfo{{Id: "rack1", DataNodeInfos: []*master_pb.DataNodeInfo{
+				{Id: "10.0.0.1:8080", DiskInfos: disk(&master_pb.VolumeEcShardInformationMessage{Id: 100, Collection: "col1", EcIndexBits: 0x3FFF})},
+				{Id: "10.0.0.2:8080", DiskInfos: disk(loaded...)},
+				{Id: "10.0.0.3:8080", DiskInfos: disk()},
+				{Id: "10.0.0.4:8080", DiskInfos: disk()},
+			}}},
+		}},
+	}
+	at := topology.NewActiveTopology(0)
+	if err := at.UpdateTopology(topoInfo); err != nil {
+		t.Fatalf("UpdateTopology: %v", err)
+	}
+	clusterInfo := &types.ClusterInfo{ActiveTopology: at}
+
+	all, _, err := Detection(context.Background(), nil, clusterInfo, NewDefaultConfig(), 0)
+	if err != nil {
+		t.Fatalf("Detection: %v", err)
+	}
+	withinRack := 0
+	for _, r := range all {
+		if strings.Contains(r.Reason, " within_rack:") {
+			withinRack++
+		}
+	}
+	if withinRack < 2 {
+		t.Fatalf("fixture plans %d within-rack moves, want at least 2", withinRack)
+	}
+
+	results, hasMore, err := Detection(context.Background(), nil, clusterInfo, NewDefaultConfig(), 1)
+	if err != nil {
+		t.Fatalf("Detection: %v", err)
+	}
+	if len(results) != 1 || !hasMore {
+		t.Fatalf("got %d results (hasMore %v), want 1 and more", len(results), hasMore)
+	}
+	if !strings.Contains(results[0].Reason, " global:") {
+		t.Errorf("kept %q, want a global move", results[0].Reason)
+	}
+}