Fix MigratingSlot solve the cluster has been updated by others

consts/context_key.go

@@ -30,3 +30,7 @@ const (
 	HeaderIsRedirect           = "X-Is-Redirect"
 	HeaderDontCheckClusterMode = "X-Dont-Check-Cluster-Mode"
 )
+
+const (
+	MigrateFailMaxRetry = 3
+)

consts/errors.go

@@ -38,5 +38,5 @@ var (
 	ErrShardIsServicing                 = errors.New("shard is servicing")
 	ErrShardSlotIsMigrating             = errors.New("shard slot is migrating")
 	ErrShardNoMatchNewMaster            = errors.New("no match new master in shard")
-	ErrSlotStartAndStopEqual            = errors.New("start and stop of a range cannot be equal")
+	ErrClusterUpdatedByOthers           = errors.New("the cluster has been updated by others")
 )

server/api/cluster.go

@@ -153,10 +153,44 @@ func (handler *ClusterHandler) MigrateSlot(c *gin.Context) {
 		return
 	}
 
-	err = handler.s.UpdateCluster(c, namespace, cluster)
-	if err != nil {
-		helper.ResponseError(c, err)
-		return
+	if err = handler.s.UpdateCluster(c, namespace, cluster); err != nil {
+		if !errors.Is(err, consts.ErrClusterUpdatedByOthers) {
+			helper.ResponseError(c, err)
+			return
+		}
+
+		retryCount := 0
+		for ; retryCount < consts.MigrateFailMaxRetry; retryCount++ {
+			newCluster, getErr := s.GetCluster(c, namespace, clusterName)
+			if getErr != nil {
+				err = getErr
+				continue
+			}
+
+			sourceShardIdx, findErr := newCluster.FindShardIndexBySlot(req.Slot)
+			if findErr != nil {
+				err = findErr
+				continue
+			}
+
+			newCluster.Shards[sourceShardIdx].MigratingSlot = &store.MigratingSlot{
+				SlotRange:   req.Slot,
+				IsMigrating: true,
+			}
+			newCluster.Shards[sourceShardIdx].TargetShardIndex = req.Target
+
+			err = handler.s.UpdateCluster(c, namespace, newCluster)
+
+			if !errors.Is(err, consts.ErrClusterUpdatedByOthers) {
+				helper.ResponseError(c, err)
+				return
+			}
+		}
+
+		if err != nil {
+			helper.ResponseError(c, err)
+			return
+		}
 	}
 	helper.ResponseOK(c, gin.H{"cluster": cluster})
 }

store/cluster.go

@@ -176,7 +176,7 @@ func (cluster *Cluster) Reset(ctx context.Context) error {
 	return nil
 }
 
-func (cluster *Cluster) findShardIndexBySlot(slot SlotRange) (int, error) {
+func (cluster *Cluster) FindShardIndexBySlot(slot SlotRange) (int, error) {
 	sourceShardIdx := -1
 	for i := 0; i < len(cluster.Shards); i++ {
 		slotRanges := cluster.Shards[i].SlotRanges
@@ -199,7 +199,7 @@ func (cluster *Cluster) MigrateSlot(ctx context.Context, slot SlotRange, targetS
 	if targetShardIdx < 0 || targetShardIdx >= len(cluster.Shards) {
 		return consts.ErrIndexOutOfRange
 	}
-	sourceShardIdx, err := cluster.findShardIndexBySlot(slot)
+	sourceShardIdx, err := cluster.FindShardIndexBySlot(slot)
 	if err != nil {
 		return err
 	}

store/cluster_test.go

@@ -44,19 +44,19 @@ func TestCluster_FindIndexShardBySlot(t *testing.T) {
 
 	slotRange, err := NewSlotRange(0, 0)
 	require.NoError(t, err)
-	shard, err := cluster.findShardIndexBySlot(slotRange)
+	shard, err := cluster.FindShardIndexBySlot(slotRange)
 	require.NoError(t, err)
 	require.Equal(t, 0, shard)
 
 	slotRange, err = NewSlotRange(MaxSlotID/3+1, MaxSlotID/3+1)
 	require.NoError(t, err)
-	shard, err = cluster.findShardIndexBySlot(slotRange)
+	shard, err = cluster.FindShardIndexBySlot(slotRange)
 	require.NoError(t, err)
 	require.Equal(t, 1, shard)
 
 	slotRange, err = NewSlotRange(MaxSlotID, MaxSlotID)
 	require.NoError(t, err)
-	shard, err = cluster.findShardIndexBySlot(slotRange)
+	shard, err = cluster.FindShardIndexBySlot(slotRange)
 	require.NoError(t, err)
 	require.Equal(t, 2, shard)
 }