Force-apply topology on reconcile and add a per-cluster sync endpoint (#395)

controller/cluster.go

@@ -117,22 +117,22 @@ func (c *ClusterChecker) WithFailoverOptions(opts store.FailoverOptions) *Cluste
 	return c
 }
 
-func (c *ClusterChecker) probeNode(ctx context.Context, node store.Node) (int64, error) {
+func (c *ClusterChecker) probeNode(ctx context.Context, node store.Node) (*store.ClusterInfo, error) {
 	clusterInfo, err := node.GetClusterInfo(ctx)
 	if err != nil {
 		// We need to use the string contains to check the error message
 		// since Kvrocks wrongly returns the error message with `ERR` prefix.
 		// And it's fixed in PR: https://github.com/apache/kvrocks/pull/2362,
 		// but we need to be compatible with the old version here.
 		if strings.Contains(err.Error(), ErrRestoringBackUp.Error()) {
-			return -1, ErrRestoringBackUp
+			return nil, ErrRestoringBackUp
 		} else if strings.Contains(err.Error(), ErrClusterNotInitialized.Error()) {
-			return -1, ErrClusterNotInitialized
+			return nil, ErrClusterNotInitialized
 		} else {
-			return -1, err
+			return nil, err
 		}
 	}
-	return clusterInfo.CurrentEpoch, nil
+	return clusterInfo, nil
 }
 
 func (c *ClusterChecker) increaseFailureCount(shardIndex int, node store.Node) int64 {
@@ -230,8 +230,9 @@ func (c *ClusterChecker) syncClusterToNodes(ctx context.Context) error {
 					zap.Int64("version", version),
 					zap.String("node_id", n.ID()),
 					zap.String("addr", n.Addr()))
-				// sync the clusterName to the latest version
-				if err := n.SyncClusterInfo(ctx, clusterInfo); err != nil {
+				// sync the clusterName to the latest version; force so a drifted-but-equal-version
+				// node is actually repaired rather than no-op'd by the server's version gate.
+				if err := n.SyncClusterInfo(ctx, clusterInfo, true); err != nil {
 					log.Error("Failed to sync the cluster topology to the node", zap.Error(err))
 				} else {
 					log.Info("Succeed to sync the cluster topology to the node")
@@ -242,12 +243,38 @@ func (c *ClusterChecker) syncClusterToNodes(ctx context.Context) error {
 	return nil
 }
 
+// topologyDiverged reports whether a probed node's applied topology disagrees with the desired one:
+// a wrong peer count or wrong slot coverage. It returns false when the node did not report these
+// fields (KnownNodes/SlotsOk < 0), so an older kvrocks that omits them falls back to epoch-only
+// reconcile instead of being force-pushed on every tick.
+func topologyDiverged(info *store.ClusterInfo, wantNodes, wantSlots int64) bool {
+	if info == nil || info.KnownNodes < 0 || info.SlotsOk < 0 {
+		return false
+	}
+	return info.KnownNodes != wantNodes || info.SlotsOk != wantSlots
+}
+
 func (c *ClusterChecker) parallelProbeNodes(ctx context.Context, cluster *store.Cluster) {
 	var mu sync.Mutex
 	var latestNodeVersion int64 = 0
 	var latestClusterNodesStr string
 	var wg sync.WaitGroup
 
+	// Desired peer count and covered-slot count from the stored (authoritative) topology. Comparing a
+	// node's applied view against these detects drift at an equal epoch; using the DESIRED coverage
+	// (not a hardcoded 16384) avoids false drift while a cluster is intentionally mid-scale with some
+	// slots not yet assigned.
+	wantNodes := int64(0)
+	wantSlots := int64(0)
+	for _, shard := range cluster.Shards {
+		wantNodes += int64(len(shard.Nodes))
+		for _, sr := range shard.SlotRanges {
+			if sr.Start >= 0 && sr.Stop >= sr.Start {
+				wantSlots += int64(sr.Stop - sr.Start + 1)
+			}
+		}
+	}
+
 	for i, shard := range cluster.Shards {
 		for _, node := range shard.Nodes {
 			wg.Add(1)
@@ -259,7 +286,7 @@ func (c *ClusterChecker) parallelProbeNodes(ctx context.Context, cluster *store.
 					zap.Bool("is_master", n.IsMaster()),
 					zap.String("addr", n.Addr()),
 				)
-				version, err := c.probeNode(ctx, n)
+				info, err := c.probeNode(ctx, n)
 				// Don't sync the cluster info to the node if it is restoring the db from backup
 				if errors.Is(err, ErrRestoringBackUp) {
 					log.Error("The node is restoring the db from backup")
@@ -274,10 +301,20 @@ func (c *ClusterChecker) parallelProbeNodes(ctx context.Context, cluster *store.
 				}
 				log.Debug("Probe the clusterName node")
 
+				// An uninitialized node (nil info) reports version -1, so the "behind" test below pushes it.
+				var version int64 = -1
+				if info != nil {
+					version = info.CurrentEpoch
+				}
+
 				clusterVersion := cluster.Version.Load()
-				if version < clusterVersion {
-					// sync the clusterName to the latest version
-					if err := n.SyncClusterInfo(ctx, cluster); err != nil {
+				// Push when the node is behind the store, OR when it matches the epoch but its applied
+				// topology has drifted (a dropped/partial push the epoch comparison alone cannot see —
+				// issue #395). Force so an equal-version repair is actually applied: the server no-ops an
+				// equal-version SETNODES otherwise. A forced push replaces the node's whole topology, so
+				// it also clears phantom/empty-address entries without a data-destroying CLUSTER RESET.
+				if version < clusterVersion || (version == clusterVersion && topologyDiverged(info, wantNodes, wantSlots)) {
+					if err := n.SyncClusterInfo(ctx, cluster, true); err != nil {
 						log.With(zap.Error(err)).Error("Failed to sync the clusterName info")
 					}
 				} else if version > clusterVersion {

controller/cluster_test.go

@@ -307,3 +307,113 @@ func TestCluster_MigrateSlot(t *testing.T) {
 	defer ticker.Stop()
 	<-ticker.C
 }
+
+// TestCluster_ReconcileForcePush drives the real reconcile path and asserts when it pushes. The
+// equal-epoch-but-diverged case is the regression guard: the previous epoch-only logic re-pushed only
+// when a node's version was strictly behind, so a node drifted at an equal epoch was never repaired.
+func TestCluster_ReconcileForcePush(t *testing.T) {
+	ctx := context.Background()
+	ns, clusterName := "test-ns", "reconcile"
+
+	newChecker := func() (*ClusterChecker, *store.Cluster, []*store.ClusterMockNode) {
+		s := NewMockClusterStore()
+		nodes := make([]*store.ClusterMockNode, 3)
+		for i := range nodes {
+			nodes[i] = store.NewClusterMockNode()
+			nodes[i].SetRole(store.RoleSlave)
+		}
+		nodes[0].SetRole(store.RoleMaster)
+		cluster := &store.Cluster{
+			Name: clusterName,
+			Shards: []*store.Shard{{
+				Nodes:            []store.Node{nodes[0], nodes[1], nodes[2]},
+				SlotRanges:       []store.SlotRange{{Start: 0, Stop: 16383}},
+				MigratingSlot:    &store.MigratingSlot{IsMigrating: false},
+				TargetShardIndex: -1,
+			}},
+		}
+		cluster.Version.Store(1)
+		require.NoError(t, s.CreateCluster(ctx, ns, cluster))
+		c := &ClusterChecker{
+			clusterStore:  s,
+			namespace:     ns,
+			clusterName:   clusterName,
+			options:       ClusterCheckOptions{pingInterval: time.Second, maxFailureCount: 3},
+			failureCounts: make(map[string]int64),
+			syncCh:        make(chan struct{}, 1),
+		}
+		return c, cluster, nodes
+	}
+
+	forcedPushes := func(nodes []*store.ClusterMockNode) int {
+		n := 0
+		for _, node := range nodes {
+			for _, force := range node.SyncForceCalls {
+				require.True(t, force, "reconcile must push with force=true")
+				n++
+			}
+		}
+		return n
+	}
+
+	t.Run("converged: no push", func(t *testing.T) {
+		c, cluster, nodes := newChecker()
+		for _, node := range nodes {
+			node.MockClusterInfo = &store.ClusterInfo{CurrentEpoch: 1, KnownNodes: 3, SlotsOk: 16384}
+		}
+		c.parallelProbeNodes(ctx, cluster)
+		require.Equal(t, 0, forcedPushes(nodes))
+	})
+
+	t.Run("equal epoch but incomplete coverage: force push (regression)", func(t *testing.T) {
+		c, cluster, nodes := newChecker()
+		for _, node := range nodes {
+			node.MockClusterInfo = &store.ClusterInfo{CurrentEpoch: 1, KnownNodes: 3, SlotsOk: 16000}
+		}
+		c.parallelProbeNodes(ctx, cluster)
+		require.Equal(t, 3, forcedPushes(nodes))
+	})
+
+	t.Run("equal epoch but wrong peer count: force push (regression)", func(t *testing.T) {
+		c, cluster, nodes := newChecker()
+		for _, node := range nodes {
+			node.MockClusterInfo = &store.ClusterInfo{CurrentEpoch: 1, KnownNodes: 2, SlotsOk: 16384}
+		}
+		c.parallelProbeNodes(ctx, cluster)
+		require.Equal(t, 3, forcedPushes(nodes))
+	})
+
+	t.Run("behind epoch: force push", func(t *testing.T) {
+		c, cluster, nodes := newChecker()
+		for _, node := range nodes {
+			node.MockClusterInfo = &store.ClusterInfo{CurrentEpoch: 0, KnownNodes: 3, SlotsOk: 16384}
+		}
+		c.parallelProbeNodes(ctx, cluster)
+		require.Equal(t, 3, forcedPushes(nodes))
+	})
+
+	t.Run("fields unreported at equal epoch: no push", func(t *testing.T) {
+		c, cluster, nodes := newChecker()
+		for _, node := range nodes {
+			node.MockClusterInfo = &store.ClusterInfo{CurrentEpoch: 1, KnownNodes: -1, SlotsOk: -1}
+		}
+		c.parallelProbeNodes(ctx, cluster)
+		require.Equal(t, 0, forcedPushes(nodes))
+	})
+}
+
+func TestTopologyDiverged(t *testing.T) {
+	const wantNodes, wantSlots = int64(6), int64(16384)
+
+	// Converged: peer count and coverage both match desired.
+	require.False(t, topologyDiverged(&store.ClusterInfo{KnownNodes: 6, SlotsOk: 16384}, wantNodes, wantSlots))
+	// Wrong peer count (e.g. a phantom or missing node).
+	require.True(t, topologyDiverged(&store.ClusterInfo{KnownNodes: 7, SlotsOk: 16384}, wantNodes, wantSlots))
+	// Incomplete coverage.
+	require.True(t, topologyDiverged(&store.ClusterInfo{KnownNodes: 6, SlotsOk: 16000}, wantNodes, wantSlots))
+	// Fields not reported (older kvrocks): fall back to epoch-only, never treated as diverged.
+	require.False(t, topologyDiverged(&store.ClusterInfo{KnownNodes: -1, SlotsOk: -1}, wantNodes, wantSlots))
+	require.False(t, topologyDiverged(nil, wantNodes, wantSlots))
+	// A cluster intentionally mid-scale (partial desired coverage) is converged when the node matches.
+	require.False(t, topologyDiverged(&store.ClusterInfo{KnownNodes: 6, SlotsOk: 8192}, wantNodes, 8192))
+}

server/api/cluster.go

@@ -161,6 +161,49 @@ func (handler *ClusterHandler) MigrateSlot(c *gin.Context) {
 	helper.ResponseOK(c, gin.H{"cluster": cluster})
 }
 
+// Sync force-pushes the stored (authoritative) topology to every node in the cluster and reports
+// which nodes accepted it. Unlike the probe loop — which only re-pushes a node whose epoch is behind
+// — this re-asserts the full topology unconditionally, so it repairs nodes that have drifted at an
+// equal epoch or hold phantom/empty-address entries (apache/kvrocks-controller#395). It performs no
+// CLUSTER RESET and never wipes data, so it is safe to run on a populated cluster.
+func (handler *ClusterHandler) Sync(c *gin.Context) {
+	namespace := c.Param("namespace")
+	clusterName := c.Param("cluster")
+
+	lock := handler.getLock(namespace, clusterName)
+	lock.Lock()
+	defer lock.Unlock()
+
+	cluster, err := handler.s.GetCluster(c, namespace, clusterName)
+	if err != nil {
+		helper.ResponseError(c, err)
+		return
+	}
+
+	synced := make([]string, 0)
+	failures := make(map[string]string)
+	for _, shard := range cluster.Shards {
+		for _, node := range shard.Nodes {
+			// Skip nodes the controller already knows are down (e.g. a drained replica); pushing to
+			// them would only add expected errors to the report.
+			if node.Failed() {
+				continue
+			}
+			if err := node.SyncClusterInfo(c, cluster, true); err != nil {
+				failures[node.Addr()] = err.Error()
+			} else {
+				synced = append(synced, node.Addr())
+			}
+		}
+	}
+
+	if len(failures) > 0 {
+		helper.ResponseError(c, fmt.Errorf("synced %d node(s); %d failed: %v", len(synced), len(failures), failures))
+		return
+	}
+	helper.ResponseOK(c, gin.H{"synced": synced, "version": cluster.Version.Load()})
+}
+
 func (handler *ClusterHandler) Import(c *gin.Context) {
 	namespace := c.Param("namespace")
 	clusterName := c.Param("cluster")

server/api/cluster_test.go

@@ -23,6 +23,7 @@ import (
 	"bytes"
 	"context"
 	"encoding/json"
+	"errors"
 	"io"
 	"net/http"
 	"net/http/httptest"
@@ -180,7 +181,7 @@ func TestClusterImport(t *testing.T) {
 	require.NoError(t, err)
 	ctx := context.Background()
 	require.NoError(t, cluster.Reset(ctx))
-	require.NoError(t, clusterNode.SyncClusterInfo(ctx, cluster))
+	require.NoError(t, clusterNode.SyncClusterInfo(ctx, cluster, true))
 	defer func() {
 		// clean up the cluster information to avoid affecting other tests
 		require.NoError(t, cluster.Reset(ctx))
@@ -303,3 +304,62 @@ func TestClusterMigrateData(t *testing.T) {
 		}
 	})
 }
+
+// syncFakeStore returns a fixed cluster (with mock nodes) so the Sync handler can be tested without
+// serializing through the storage engine, which would turn mock nodes back into real ClusterNodes.
+type syncFakeStore struct {
+	store.Store
+	cluster *store.Cluster
+}
+
+func (s *syncFakeStore) GetCluster(ctx context.Context, ns, name string) (*store.Cluster, error) {
+	if s.cluster == nil {
+		return nil, consts.ErrNotFound
+	}
+	return s.cluster, nil
+}
+
+func TestClusterHandler_Sync(t *testing.T) {
+	ns, name := "test-ns", "sync-cluster"
+	full := []store.SlotRange{{Start: 0, Stop: 16383}}
+
+	newHandler := func(nodes []store.Node) *ClusterHandler {
+		cluster := &store.Cluster{Name: name, Shards: store.Shards{{Nodes: nodes, SlotRanges: full}}}
+		cluster.Version.Store(1)
+		return &ClusterHandler{s: &syncFakeStore{Store: store.NewClusterStore(engine.NewMock()), cluster: cluster}}
+	}
+	call := func(h *ClusterHandler) *httptest.ResponseRecorder {
+		recorder := httptest.NewRecorder()
+		ctx := GetTestContext(recorder)
+		ctx.Params = []gin.Param{{Key: "namespace", Value: ns}, {Key: "cluster", Value: name}}
+		h.Sync(ctx)
+		return recorder
+	}
+
+	t.Run("force-pushes healthy nodes and skips failed", func(t *testing.T) {
+		n0 := store.NewClusterMockNode()
+		n0.SetRole(store.RoleMaster)
+		n1 := store.NewClusterMockNode()
+		n1.SetRole(store.RoleSlave)
+		n2 := store.NewClusterMockNode()
+		n2.SetRole(store.RoleSlave)
+		n2.SetStatus(store.NodeStatusFailed)
+
+		rec := call(newHandler([]store.Node{n0, n1, n2}))
+		require.Equal(t, http.StatusOK, rec.Code)
+		require.Equal(t, []bool{true}, n0.SyncForceCalls, "healthy node force-pushed")
+		require.Equal(t, []bool{true}, n1.SyncForceCalls, "healthy node force-pushed")
+		require.Empty(t, n2.SyncForceCalls, "failed node must be skipped")
+	})
+
+	t.Run("reports push failures with a non-OK status", func(t *testing.T) {
+		n0 := store.NewClusterMockNode()
+		n0.SetRole(store.RoleMaster)
+		n1 := store.NewClusterMockNode()
+		n1.SetRole(store.RoleSlave)
+		n1.SyncErr = errors.New("push failed")
+
+		rec := call(newHandler([]store.Node{n0, n1}))
+		require.NotEqual(t, http.StatusOK, rec.Code)
+	})
+}

server/api/shard.go

@@ -192,13 +192,13 @@ func (handler *ShardHandler) Failover(c *gin.Context) {
 	wg.Add(2)
 	go func() {
 		defer wg.Done()
-		if e := oldMaster.SyncClusterInfo(c, cluster); e != nil {
+		if e := oldMaster.SyncClusterInfo(c, cluster, true); e != nil {
 			logger.Get().With(zap.Error(e), zap.String("node", oldMaster.Addr())).Warn("Failed to sync cluster info to old master")
 		}
 	}()
 	go func() {
 		defer wg.Done()
-		if e := newMaster.SyncClusterInfo(c, cluster); e != nil {
+		if e := newMaster.SyncClusterInfo(c, cluster, true); e != nil {
 			logger.Get().With(zap.Error(e), zap.String("node", newMaster.Addr())).Warn("Failed to sync cluster info to new master")
 		}
 	}()

server/api/shard_test.go

@@ -258,8 +258,8 @@ func TestClusterFailover(t *testing.T) {
 		}
 
 		// sync cluster info to each node
-		require.NoError(t, node0.SyncClusterInfo(ctx, gotCluster))
-		require.NoError(t, node1.SyncClusterInfo(ctx, gotCluster))
+		require.NoError(t, node0.SyncClusterInfo(ctx, gotCluster, true))
+		require.NoError(t, node1.SyncClusterInfo(ctx, gotCluster, true))
 
 		clusterNodeInfo0, err := node0.GetClusterNodeInfo(ctx)
 		require.NoError(t, err)

server/route.go

@@ -71,6 +71,7 @@ func (srv *Server) initHandlers() {
 			clusters.GET("/:cluster", middleware.RequiredCluster, handler.Cluster.Get)
 			clusters.DELETE("/:cluster", middleware.RequiredCluster, handler.Cluster.Remove)
 			clusters.POST("/:cluster/migrate", handler.Cluster.MigrateSlot)
+			clusters.POST("/:cluster/sync", middleware.RequiredCluster, handler.Cluster.Sync)
 			clusters.POST("/:cluster/nodes/status", middleware.RequiredCluster, handler.Node.SetStatus)
 		}