volume.fix.replication: add a well-placed replica before deleting a misplaced one (#10364)

* volume.fix.replication: add a well-placed replica before deleting a misplaced one A misplaced volume with no surplus replica (replica count == copy count) used to have its misplaced replica deleted first, and the replacement copied only on the next pass. That drops the volume below its intended durability, and permanently so when no destination can accept the replacement copy. Now such volumes get a well-placed replica first, and the misplaced one is trimmed as surplus on a later pass, following the same fulfill-before-delete principle as volume.tier.move (#8950). The classification switch is extracted into classifyReplicaSet so the add-wins-over-trim priority is unit-testable. Also stop the fix loop when an -apply pass makes no progress (nothing copied, nothing deleted); previously an unplaceable under-replicated volume made the loop spin forever, re-collecting the topology every 15 seconds. And propagate ewg.Wait() errors instead of swallowing them. * volume.fix.replication: drop unused allLocations from deleteOneVolume --------- Co-authored-by: Chris Lu <chris.lu@gmail.com>
2026-07-24 17:12:54 +00:00 · 2026-07-21 12:43:38 -07:00
parent 742bd9f3aa
commit ca06589d64
2 changed files with 162 additions and 10 deletions
@@ -7,6 +7,7 @@ import (
 	"strconv"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"time"

 	"slices"
@@ -43,6 +44,10 @@ func (c *commandVolumeFixReplication) Help() string {
 	This command also finds all under-replicated volumes, and finds volume servers with free slots.
 	If the free slots satisfy the replication requirement, the volume content is copied over and mounted.

+	Misplaced volumes with a surplus replica have a misplaced replica deleted. Without a surplus, a
+	well-placed replica is added first and the misplaced one is trimmed on a later pass, so the volume
+	never drops below its intended replica count.
+
 	volume.fix.replication                                # do not take action
 	volume.fix.replication -apply                         # actually deleting or copying the volume files and mount the volume
 	volume.fix.replication -collectionPattern=important*  # fix any collections with prefix "important"
@@ -137,14 +142,17 @@ func (c *commandVolumeFixReplication) Do(args []string, commandEnv *CommandEnv,
 				fmt.Fprintf(writer, "checking volume %d replication %s has %d replicas [%s]\n", replica.info.Id, replicaPlacement, len(replicas), strings.Join(locations, ", "))
 			}

-			switch {
-			case replicaPlacement.GetCopyCount() > len(replicas) || !satisfyReplicaCurrentLocation(replicaPlacement, replicas):
+			switch classifyReplicaSet(replicaPlacement, replicas) {
+			case replicaFixAddOne:
 				underReplicatedVolumeIds = append(underReplicatedVolumeIds, vid)
 				fmt.Fprintf(writer, "volume %d replication %s, but under replicated %+d\n", replica.info.Id, replicaPlacement, len(replicas))
-			case isMisplaced(replicas, replicaPlacement):
+			case replicaFixAddOneBeforeTrim:
+				underReplicatedVolumeIds = append(underReplicatedVolumeIds, vid)
+				fmt.Fprintf(writer, "volume %d replication %s is not well placed [%s], adding a well-placed replica before trimming the misplaced one\n", replica.info.Id, replicaPlacement, strings.Join(locations, ", "))
+			case replicaFixTrimMisplaced:
 				misplacedVolumeIds = append(misplacedVolumeIds, vid)
 				fmt.Fprintf(writer, "volume %d replication %s is not well placed [%s]\n", replica.info.Id, replicaPlacement, strings.Join(locations, ", "))
-			case replicaPlacement.GetCopyCount() < len(replicas):
+			case replicaFixTrimOver:
 				overReplicatedVolumeIds = append(overReplicatedVolumeIds, vid)
 				fmt.Fprintf(writer, "volume %d replication %s, but over replicated %+d\n", replica.info.Id, replicaPlacement, len(replicas))
 			}
@@ -155,6 +163,7 @@ func (c *commandVolumeFixReplication) Do(args []string, commandEnv *CommandEnv,
 			return fmt.Errorf("lock is lost")
 		}

+		var deletedVolumeReplicas atomic.Int64
 		ewg.Reset()
 		ewg.Add(func() error {
 			// find the most underpopulated data nodes
@@ -163,14 +172,18 @@ func (c *commandVolumeFixReplication) Do(args []string, commandEnv *CommandEnv,
 		})
 		if *doDelete {
 			ewg.Add(func() error {
-				return c.deleteOneVolume(commandEnv, writer, *applyChanges, *doCheck, overReplicatedVolumeIds, volumeReplicas, allLocations, pickOneReplicaToDelete)
+				deleted, err := c.deleteOneVolume(commandEnv, writer, *applyChanges, *doCheck, overReplicatedVolumeIds, volumeReplicas, pickOneReplicaToDelete)
+				deletedVolumeReplicas.Add(int64(deleted))
+				return err
 			})
 			ewg.Add(func() error {
-				return c.deleteOneVolume(commandEnv, writer, *applyChanges, *doCheck, misplacedVolumeIds, volumeReplicas, allLocations, pickOneMisplacedVolume)
+				deleted, err := c.deleteOneVolume(commandEnv, writer, *applyChanges, *doCheck, misplacedVolumeIds, volumeReplicas, pickOneMisplacedVolume)
+				deletedVolumeReplicas.Add(int64(deleted))
+				return err
 			})
 		}
 		if err := ewg.Wait(); err != nil {
-			return nil
+			return err
 		}

 		if !*applyChanges {
@@ -206,10 +219,60 @@ func (c *commandVolumeFixReplication) Do(args []string, commandEnv *CommandEnv,
 				}
 			}
 		}
+
+		// Without progress the next pass would reclassify the same volumes and
+		// loop forever, e.g. when no destination can accept a replica; stop
+		// instead. Deletions count as progress since they free up slots.
+		if underReplicatedVolumeIdsCount > 0 && len(fixedVolumeReplicas) == 0 && deletedVolumeReplicas.Load() == 0 {
+			fmt.Fprintf(writer, "no progress made on %d under replicated volumes, stopping; free up capacity or adjust replica placement, then re-run\n", underReplicatedVolumeIdsCount)
+			break
+		}
 	}
 	return nil
 }

+// replicaFix is the single next action volume.fix.replication takes on one
+// volume's replica set.
+type replicaFix int
+
+const (
+	replicaFixNothing replicaFix = iota
+	// replicaFixAddOne: the volume lacks a replica, in count or in
+	// failure-domain spread; copy one to a well-placed destination.
+	replicaFixAddOne
+	// replicaFixAddOneBeforeTrim: the replica count is complete but a replica
+	// is misplaced. Add a well-placed replica first; a later pass trims the
+	// misplaced one as surplus. Deleting first would drop the volume below its
+	// intended durability — for good, if no destination can take the
+	// replacement copy.
+	replicaFixAddOneBeforeTrim
+	// replicaFixTrimMisplaced: more replicas than the policy asks for, at
+	// least one misplaced; delete a misplaced one.
+	replicaFixTrimMisplaced
+	// replicaFixTrimOver: more replicas than the policy asks for; delete a
+	// surplus one.
+	replicaFixTrimOver
+)
+
+// classifyReplicaSet decides the next action for one volume. Add always wins
+// over trim: a volume that lacks a replica and also has a misplaced or surplus
+// one gets its missing replica first, and the trim happens on a later pass
+// once the new copy registered in the topology.
+func classifyReplicaSet(replicaPlacement *super_block.ReplicaPlacement, replicas []*VolumeReplica) replicaFix {
+	switch {
+	case replicaPlacement.GetCopyCount() > len(replicas) || !satisfyReplicaCurrentLocation(replicaPlacement, replicas):
+		return replicaFixAddOne
+	case isMisplaced(replicas, replicaPlacement):
+		if len(replicas) <= replicaPlacement.GetCopyCount() {
+			return replicaFixAddOneBeforeTrim
+		}
+		return replicaFixTrimMisplaced
+	case replicaPlacement.GetCopyCount() < len(replicas):
+		return replicaFixTrimOver
+	}
+	return replicaFixNothing
+}
+
 func collectVolumeReplicaLocations(topologyInfo *master_pb.TopologyInfo) (map[uint32][]*VolumeReplica, []location) {
 	volumeReplicas := make(map[uint32][]*VolumeReplica)
 	var allLocations []location
@@ -278,10 +341,12 @@ func (c *commandVolumeFixReplication) matchCollectionPattern(collection string)
 	return wildcard.MatchesWildcard(*c.collectionPattern, collection)
 }

-func (c *commandVolumeFixReplication) deleteOneVolume(commandEnv *CommandEnv, writer io.Writer, applyChanges bool, doCheck bool, volumeIds []uint32, volumeReplicas map[uint32][]*VolumeReplica, allLocations []location, selectOneVolumeFn SelectOneVolumeFunc) error {
+// deleteOneVolume trims one replica from each of the given volumes, and
+// reports how many replicas it actually deleted.
+func (c *commandVolumeFixReplication) deleteOneVolume(commandEnv *CommandEnv, writer io.Writer, applyChanges bool, doCheck bool, volumeIds []uint32, volumeReplicas map[uint32][]*VolumeReplica, selectOneVolumeFn SelectOneVolumeFunc) (deleted int, err error) {
 	if len(volumeIds) == 0 {
 		// nothing to do
-		return nil
+		return 0, nil
 	}

 	for _, vid := range volumeIds {
@@ -334,10 +399,12 @@ func (c *commandVolumeFixReplication) deleteOneVolume(commandEnv *CommandEnv, wr
 		if err := deleteVolume(commandEnv.option.GrpcDialOption, needle.VolumeId(replica.info.Id),
 			pb.NewServerAddressFromDataNode(replica.location.dataNode), false, true); err != nil {
 			fmt.Fprintf(writer, "deleting volume %d from %s : %v", replica.info.Id, replica.location.dataNode.Id, err)
+		} else {
+			deleted++
 		}

 	}
-	return nil
+	return deleted, nil
 }

 func (c *commandVolumeFixReplication) fixUnderReplicatedVolumes(commandEnv *CommandEnv, writer io.Writer, applyChanges bool, volumeIds []uint32, volumeReplicas map[uint32][]*VolumeReplica, allLocations []location, retryCount int, volumesPerStep int, maxParallelization int, maxParallelizationPerServer int) (fixedVolumes map[string]int, err error) {
@@ -646,3 +646,88 @@ func TestSatisfyReplicaCurrentLocation(t *testing.T) {
 		})
 	}
 }
+
+func TestClassifyReplicaSet(t *testing.T) {
+
+	var tests = []struct {
+		name        string
+		replication string
+		replicas    []*VolumeReplica
+		expected    replicaFix
+	}{
+		{
+			// The full replica count is present but two copies crowd one rack.
+			// Deleting a misplaced replica first would drop the volume below its
+			// intended durability, so add a well-placed replica first.
+			name:        "110 misplaced without surplus adds before trimming",
+			replication: "110",
+			replicas: []*VolumeReplica{
+				{location: &location{"dc1", "r1", &master_pb.DataNodeInfo{Id: "dn1"}}},
+				{location: &location{"dc1", "r1", &master_pb.DataNodeInfo{Id: "dn2"}}},
+				{location: &location{"dc2", "r2", &master_pb.DataNodeInfo{Id: "dn3"}}},
+			},
+			expected: replicaFixAddOneBeforeTrim,
+		},
+		{
+			// Same misplaced set once the well-placed replica registered: now
+			// there is a surplus, and the misplaced replica can be trimmed.
+			name:        "110 misplaced with surplus trims",
+			replication: "110",
+			replicas: []*VolumeReplica{
+				{location: &location{"dc1", "r1", &master_pb.DataNodeInfo{Id: "dn1"}}},
+				{location: &location{"dc1", "r1", &master_pb.DataNodeInfo{Id: "dn2"}}},
+				{location: &location{"dc1", "r3", &master_pb.DataNodeInfo{Id: "dn4"}}},
+				{location: &location{"dc2", "r2", &master_pb.DataNodeInfo{Id: "dn3"}}},
+			},
+			expected: replicaFixTrimMisplaced,
+		},
+		{
+			name:        "010 missing a replica adds",
+			replication: "010",
+			replicas: []*VolumeReplica{
+				{location: &location{"dc1", "r1", &master_pb.DataNodeInfo{Id: "dn1"}}},
+			},
+			expected: replicaFixAddOne,
+		},
+		{
+			// Simultaneously over-provisioned in count and lacking rack spread:
+			// the missing failure domain is filled first, trims come later.
+			name:        "010 surplus count but missing rack spread adds",
+			replication: "010",
+			replicas: []*VolumeReplica{
+				{location: &location{"dc1", "r1", &master_pb.DataNodeInfo{Id: "dn1"}}},
+				{location: &location{"dc1", "r1", &master_pb.DataNodeInfo{Id: "dn2"}}},
+				{location: &location{"dc1", "r1", &master_pb.DataNodeInfo{Id: "dn3"}}},
+			},
+			expected: replicaFixAddOne,
+		},
+		{
+			name:        "010 well placed needs nothing",
+			replication: "010",
+			replicas: []*VolumeReplica{
+				{location: &location{"dc1", "r1", &master_pb.DataNodeInfo{Id: "dn1"}}},
+				{location: &location{"dc1", "r2", &master_pb.DataNodeInfo{Id: "dn2"}}},
+			},
+			expected: replicaFixNothing,
+		},
+		{
+			name:        "000 duplicate copy trims",
+			replication: "000",
+			replicas: []*VolumeReplica{
+				{location: &location{"dc1", "r1", &master_pb.DataNodeInfo{Id: "dn1"}}},
+				{location: &location{"dc1", "r2", &master_pb.DataNodeInfo{Id: "dn2"}}},
+			},
+			expected: replicaFixTrimMisplaced,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			replicaPlacement, _ := super_block.NewReplicaPlacementFromString(tt.replication)
+			if got := classifyReplicaSet(replicaPlacement, tt.replicas); got != tt.expected {
+				t.Errorf("%s: expect %v got %v for %v %+v",
+					tt.name, tt.expected, got, tt.replication, tt.replicas)
+			}
+		})
+	}
+}