diff --git a/controller/cluster.go b/controller/cluster.go
index b6f78bc..59778d3 100755
--- a/controller/cluster.go
+++ b/controller/cluster.go
@@ -117,7 +117,7 @@ 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
@@ -125,14 +125,14 @@ func (c *ClusterChecker) probeNode(ctx context.Context, node store.Node) (int64,
 		// 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 {
diff --git a/controller/cluster_test.go b/controller/cluster_test.go
index a377444..3f55619 100644
--- a/controller/cluster_test.go
+++ b/controller/cluster_test.go
@@ -22,6 +22,7 @@ package controller
 
 import (
 	"context"
+	"errors"
 	"sync"
 	"testing"
 	"time"
@@ -307,3 +308,145 @@ 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))
+	})
+
+	t.Run("restoring from backup: skipped, not pushed", func(t *testing.T) {
+		c, cluster, nodes := newChecker()
+		for _, node := range nodes {
+			node.ClusterInfoErr = ErrRestoringBackUp
+		}
+		c.parallelProbeNodes(ctx, cluster)
+		require.Equal(t, 0, forcedPushes(nodes))
+		require.Empty(t, c.failureCounts, "a restoring node is not counted as a failure")
+	})
+
+	t.Run("unreachable node: failure counted, not pushed", func(t *testing.T) {
+		c, cluster, nodes := newChecker()
+		for _, node := range nodes {
+			node.ClusterInfoErr = errors.New("dial tcp: connection refused")
+		}
+		c.parallelProbeNodes(ctx, cluster)
+		require.Equal(t, 0, forcedPushes(nodes))
+		require.NotEmpty(t, c.failureCounts, "an unreachable node increments its failure count")
+	})
+
+	t.Run("uninitialized node (CLUSTERDOWN): force-pushed", func(t *testing.T) {
+		// probeNode maps this to ErrClusterNotInitialized, which is NOT a failure — the node falls
+		// through with nil info (version -1 < stored version) and is force-pushed to initialize it.
+		c, cluster, nodes := newChecker()
+		for _, node := range nodes {
+			node.ClusterInfoErr = ErrClusterNotInitialized
+		}
+		c.parallelProbeNodes(ctx, cluster)
+		require.Equal(t, 3, forcedPushes(nodes))
+		require.Empty(t, c.failureCounts, "an uninitialized node is not counted as a failure")
+	})
+}
+
+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))
+}
diff --git a/server/api/cluster.go b/server/api/cluster.go
index 5217889..f7330aa 100644
--- a/server/api/cluster.go
+++ b/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")
diff --git a/server/api/cluster_test.go b/server/api/cluster_test.go
index d9e1470..736d4e4 100644
--- a/server/api/cluster_test.go
+++ b/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,70 @@ 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)
+	})
+
+	t.Run("missing cluster returns an error", func(t *testing.T) {
+		// cluster == nil → the fake store's GetCluster returns ErrNotFound, exercising the
+		// handler's lookup-failure branch.
+		h := &ClusterHandler{s: &syncFakeStore{Store: store.NewClusterStore(engine.NewMock())}}
+		rec := call(h)
+		require.NotEqual(t, http.StatusOK, rec.Code)
+	})
+}
diff --git a/server/api/shard.go b/server/api/shard.go
index 502b23a..d9b70df 100644
--- a/server/api/shard.go
+++ b/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")
 		}
 	}()
diff --git a/server/api/shard_test.go b/server/api/shard_test.go
index c131558..6f8a73e 100644
--- a/server/api/shard_test.go
+++ b/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)
diff --git a/server/route.go b/server/route.go
index 113a1a0..04a22ab 100644
--- a/server/route.go
+++ b/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)
 		}
 
diff --git a/store/cluster.go b/store/cluster.go
index b402594..30474fa 100644
--- a/store/cluster.go
+++ b/store/cluster.go
@@ -148,7 +148,9 @@ func (cluster *Cluster) PromoteNewMaster(ctx context.Context,
 func (cluster *Cluster) SyncToNodes(ctx context.Context) error {
 	for i := 0; i < len(cluster.Shards); i++ {
 		for _, node := range cluster.Shards[i].Nodes {
-			if err := node.SyncClusterInfo(ctx, cluster); err != nil {
+			// force: this runs right after a topology change, so assert the new state unconditionally
+			// rather than let the server's version gate no-op an equal-version push.
+			if err := node.SyncClusterInfo(ctx, cluster, true); err != nil {
 				return err
 			}
 		}
diff --git a/store/cluster_mock_node.go b/store/cluster_mock_node.go
index 200ae93..15b5957 100644
--- a/store/cluster_mock_node.go
+++ b/store/cluster_mock_node.go
@@ -34,6 +34,17 @@ type ClusterMockNode struct {
 	MasterReplOffset uint64 // used when simulating master in GetReplicationInfo
 	SlaveOffset      uint64 // used when simulating slave offset in GetReplicationInfo
 	SlaveAddr        string // when master, slave Addr for matching; empty means use mock.Addr()
+
+	// MockClusterInfo, when set, is what GetClusterInfo returns (else a zero-value ClusterInfo).
+	MockClusterInfo *ClusterInfo
+	// ClusterInfoErr, when set, is returned by GetClusterInfo (to simulate a node that is down,
+	// restoring from backup, or otherwise unreachable during a probe). Takes precedence over
+	// MockClusterInfo.
+	ClusterInfoErr error
+	// SyncForceCalls records the `force` argument of every SyncClusterInfo call, for assertions.
+	SyncForceCalls []bool
+	// SyncErr, when set, is returned by SyncClusterInfo (to simulate a push failure).
+	SyncErr error
 }
 
 var _ Node = (*ClusterMockNode)(nil)
@@ -49,11 +60,18 @@ func (mock *ClusterMockNode) GetClusterNodeInfo(ctx context.Context) (*ClusterNo
 }
 
 func (mock *ClusterMockNode) GetClusterInfo(ctx context.Context) (*ClusterInfo, error) {
+	if mock.ClusterInfoErr != nil {
+		return nil, mock.ClusterInfoErr
+	}
+	if mock.MockClusterInfo != nil {
+		return mock.MockClusterInfo, nil
+	}
 	return &ClusterInfo{}, nil
 }
 
-func (mock *ClusterMockNode) SyncClusterInfo(ctx context.Context, cluster *Cluster) error {
-	return nil
+func (mock *ClusterMockNode) SyncClusterInfo(ctx context.Context, cluster *Cluster, force bool) error {
+	mock.SyncForceCalls = append(mock.SyncForceCalls, force)
+	return mock.SyncErr
 }
 
 func (mock *ClusterMockNode) Reset(ctx context.Context) error {
diff --git a/store/cluster_mock_node_test.go b/store/cluster_mock_node_test.go
new file mode 100644
index 0000000..efc6a86
--- /dev/null
+++ b/store/cluster_mock_node_test.go
@@ -0,0 +1,69 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+package store
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+// TestClusterMockNode covers the reconcile-test hooks added to the mock: a settable GetClusterInfo
+// reply, a record of every SyncClusterInfo force flag, and an injectable push error.
+func TestClusterMockNode(t *testing.T) {
+	ctx := context.Background()
+	mock := NewClusterMockNode()
+
+	// Default: a zero-value ClusterInfo when no override is set.
+	info, err := mock.GetClusterInfo(ctx)
+	require.NoError(t, err)
+	require.NotNil(t, info)
+	require.EqualValues(t, 0, info.CurrentEpoch)
+
+	// An injected error takes precedence and is returned (simulating an unreachable node).
+	mock.ClusterInfoErr = errors.New("node is down")
+	_, err = mock.GetClusterInfo(ctx)
+	require.ErrorIs(t, err, mock.ClusterInfoErr)
+	mock.ClusterInfoErr = nil
+
+	// The override is returned verbatim so a test can simulate a node's applied view.
+	mock.MockClusterInfo = &ClusterInfo{CurrentEpoch: 5, KnownNodes: 3, SlotsOk: 16384}
+	info, err = mock.GetClusterInfo(ctx)
+	require.NoError(t, err)
+	require.EqualValues(t, 5, info.CurrentEpoch)
+	require.EqualValues(t, 3, info.KnownNodes)
+	require.EqualValues(t, 16384, info.SlotsOk)
+
+	// SyncClusterInfo records the force flag of every call and returns the configured error.
+	cluster := &Cluster{Shards: Shards{{
+		Nodes:      []Node{mock},
+		SlotRanges: []SlotRange{{Start: 0, Stop: 16383}},
+	}}}
+	cluster.Version.Store(1)
+	require.NoError(t, mock.SyncClusterInfo(ctx, cluster, true))
+	require.NoError(t, mock.SyncClusterInfo(ctx, cluster, false))
+	require.Equal(t, []bool{true, false}, mock.SyncForceCalls)
+
+	mock.SyncErr = errors.New("simulated push failure")
+	require.ErrorIs(t, mock.SyncClusterInfo(ctx, cluster, true), mock.SyncErr)
+	require.Equal(t, []bool{true, false, true}, mock.SyncForceCalls)
+}
diff --git a/store/cluster_node.go b/store/cluster_node.go
index bce333f..4a80d81 100755
--- a/store/cluster_node.go
+++ b/store/cluster_node.go
@@ -84,7 +84,7 @@ type Node interface {
 	Reset(ctx context.Context) error
 	GetClusterNodeInfo(ctx context.Context) (*ClusterNodeInfo, error)
 	GetClusterInfo(ctx context.Context) (*ClusterInfo, error)
-	SyncClusterInfo(ctx context.Context, cluster *Cluster) error
+	SyncClusterInfo(ctx context.Context, cluster *Cluster, force bool) error
 	CheckClusterMode(ctx context.Context) (int64, error)
 	MigrateSlot(ctx context.Context, slot SlotRange, NodeID string) error
 
@@ -111,6 +111,12 @@ type ClusterInfo struct {
 	CurrentEpoch   int64          `json:"cluster_current_epoch"`
 	MigratingSlot  *MigratingSlot `json:"migrating_slot"`
 	MigratingState string         `json:"migrating_state"`
+	// KnownNodes and SlotsOk are parsed from CLUSTER INFO so the reconcile loop can detect a node
+	// whose applied topology has drifted from the desired one even at an equal epoch — the case a
+	// pure epoch comparison misses (see apache/kvrocks-controller#395). SlotsOk is the count of
+	// assigned slots (16384 when fully covered); KnownNodes is the node's peer count.
+	KnownNodes int64 `json:"cluster_known_nodes"`
+	SlotsOk    int64 `json:"cluster_slots_ok"`
 }
 
 type ClusterNodeInfo struct {
@@ -252,8 +258,13 @@ func (n *ClusterNode) GetClusterInfo(ctx context.Context) (*ClusterInfo, error)
 	if err != nil {
 		return nil, err
 	}
+	return parseClusterInfo(infoStr)
+}
 
-	clusterInfo := &ClusterInfo{CurrentEpoch: -1}
+// parseClusterInfo parses the CLUSTER INFO reply. Fields absent from the reply keep their sentinel -1
+// so a partial/older kvrocks reply never looks "converged" (all-slots-covered / expected-peer-count).
+func parseClusterInfo(infoStr string) (*ClusterInfo, error) {
+	clusterInfo := &ClusterInfo{CurrentEpoch: -1, KnownNodes: -1, SlotsOk: -1}
 	lines := strings.Split(infoStr, "\r\n")
 	for _, line := range lines {
 		fields := strings.Split(line, ":")
@@ -261,22 +272,26 @@ func (n *ClusterNode) GetClusterInfo(ctx context.Context) (*ClusterInfo, error)
 			continue
 		}
 		fields[1] = strings.TrimSpace(fields[1])
+		var err error
 		switch strings.ToLower(strings.TrimSpace(fields[0])) {
 		case "cluster_current_epoch":
 			clusterInfo.CurrentEpoch, err = strconv.ParseInt(fields[1], 10, 64)
-			if err != nil {
-				return nil, err
-			}
+		case "cluster_known_nodes":
+			clusterInfo.KnownNodes, err = strconv.ParseInt(fields[1], 10, 64)
+		case "cluster_slots_ok":
+			clusterInfo.SlotsOk, err = strconv.ParseInt(fields[1], 10, 64)
 		case "migrating_slot", "migrating_slot(s)":
 			// TODO(@git-hulk): handle multiple migrating slots
-			slotRange, err := ParseSlotRange(fields[1])
-			if err != nil {
-				return nil, err
+			var slotRange *SlotRange
+			if slotRange, err = ParseSlotRange(fields[1]); err == nil {
+				clusterInfo.MigratingSlot = FromSlotRange(*slotRange)
 			}
-			clusterInfo.MigratingSlot = FromSlotRange(*slotRange)
 		case "migrating_state":
 			clusterInfo.MigratingState = fields[1]
 		}
+		if err != nil {
+			return nil, err
+		}
 	}
 	return clusterInfo, nil
 }
@@ -315,17 +330,52 @@ func (n *ClusterNode) GetClusterNodesString(ctx context.Context) (string, error)
 	return strings.TrimRight(clusterNodesStr, "\n"), nil
 }
 
-func (n *ClusterNode) SyncClusterInfo(ctx context.Context, cluster *Cluster) error {
+// syncMaxRetries and syncRetryBaseDelay bound the per-node push retry. kvrocks nodes do not gossip,
+// so the controller is the sole topology source; a push dropped by a transient blip (node restarting,
+// network hiccup) is never re-tried by any other path until the next probe tick, leaving that node on a
+// stale/divergent view (apache/kvrocks-controller#395). A short bounded backoff closes that window.
+const (
+	syncMaxRetries     = 3
+	syncRetryBaseDelay = 150 * time.Millisecond
+)
+
+// SyncClusterInfo pushes the authoritative topology to this node via CLUSTERX SETNODEID + SETNODES.
+//
+// When force is true the SETNODES carries the `force` flag, which makes kvrocks apply the topology
+// unconditionally — bypassing its version gate (reject-if-lower, no-op-if-equal). This is what lets the
+// controller REPAIR a node that has drifted at an equal epoch (the divergence #395 describes): a normal
+// equal-version push is silently no-op'd by the server, so it can never fix a bad-but-same-version view.
+// Because SETNODES replaces the node's entire topology, a forced push also removes stale/phantom node
+// entries (e.g. empty-address ghosts) without the data-destroying CLUSTER RESET the alternative requires.
+// Safe because the controller is the single writer of topology; callers must not force a version older
+// than a node legitimately ahead of the store (the probe loop keeps the `node ahead` branch for that).
+func (n *ClusterNode) SyncClusterInfo(ctx context.Context, cluster *Cluster, force bool) error {
 	clusterStr, err := cluster.ToSlotString()
 	if err != nil {
 		return err
 	}
+	args := []interface{}{"CLUSTERX", "SETNODES", clusterStr, cluster.Version.Load()}
+	if force {
+		args = append(args, "force")
+	}
 	redisCli := n.GetClient()
-	err = redisCli.Do(ctx, "CLUSTERX", "SETNODEID", n.id).Err()
-	if err != nil {
-		return err
+	var lastErr error
+	for attempt := 0; attempt < syncMaxRetries; attempt++ {
+		if attempt > 0 {
+			select {
+			case <-ctx.Done():
+				return ctx.Err()
+			case <-time.After(syncRetryBaseDelay << (attempt - 1)):
+			}
+		}
+		if lastErr = redisCli.Do(ctx, "CLUSTERX", "SETNODEID", n.id).Err(); lastErr != nil {
+			continue
+		}
+		if lastErr = redisCli.Do(ctx, args...).Err(); lastErr == nil {
+			return nil
+		}
 	}
-	return redisCli.Do(ctx, "CLUSTERX", "SETNODES", clusterStr, cluster.Version.Load()).Err()
+	return lastErr
 }
 
 func (n *ClusterNode) Reset(ctx context.Context) error {
diff --git a/store/cluster_node_test.go b/store/cluster_node_test.go
index 41bf55f..28a99ec 100644
--- a/store/cluster_node_test.go
+++ b/store/cluster_node_test.go
@@ -57,7 +57,7 @@ func TestClusterNode(t *testing.T) {
 		}}
 
 		cluster.Version.Store(1)
-		require.NoError(t, node0.SyncClusterInfo(ctx, cluster))
+		require.NoError(t, node0.SyncClusterInfo(ctx, cluster, true))
 		clusterInfo, err := node0.GetClusterInfo(ctx)
 		require.NoError(t, err)
 		require.EqualValues(t, 1, clusterInfo.CurrentEpoch)
@@ -104,3 +104,50 @@ func TestNodeInfo_Validate(t *testing.T) {
 	node.addr = "1.2.3.4"
 	require.NoError(t, node.Validate())
 }
+
+func TestParseClusterInfo(t *testing.T) {
+	// Absent fields keep the -1 sentinel so a partial reply never looks converged.
+	info, err := parseClusterInfo("cluster_state:ok\r\ncluster_current_epoch:7\r\n")
+	require.NoError(t, err)
+	require.EqualValues(t, 7, info.CurrentEpoch)
+	require.EqualValues(t, -1, info.KnownNodes)
+	require.EqualValues(t, -1, info.SlotsOk)
+
+	info, err = parseClusterInfo("cluster_current_epoch:3\r\ncluster_known_nodes:6\r\ncluster_slots_ok:16384\r\n")
+	require.NoError(t, err)
+	require.EqualValues(t, 3, info.CurrentEpoch)
+	require.EqualValues(t, 6, info.KnownNodes)
+	require.EqualValues(t, 16384, info.SlotsOk)
+
+	// A malformed integer surfaces as an error rather than a silent zero.
+	_, err = parseClusterInfo("cluster_known_nodes:not-a-number\r\n")
+	require.Error(t, err)
+}
+
+// TestSyncClusterInfo_RetryAndCancel exercises the bounded-retry push path without a live server: the
+// node points at a port with nothing listening, so every CLUSTERX call fails and the loop runs to
+// exhaustion (returning the last error) or is short-circuited by a cancelled context during backoff.
+func TestSyncClusterInfo_RetryAndCancel(t *testing.T) {
+	newCluster := func(addr string) (*ClusterNode, *Cluster) {
+		node := NewClusterNode(addr, "")
+		node.SetRole(RoleMaster)
+		cluster := &Cluster{Shards: Shards{{
+			Nodes:      []Node{node},
+			SlotRanges: []SlotRange{{Start: 0, Stop: 16383}},
+		}}}
+		cluster.Version.Store(1)
+		return node, cluster
+	}
+
+	t.Run("retries then returns the last error", func(t *testing.T) {
+		node, cluster := newCluster("127.0.0.1:6399")
+		require.Error(t, node.SyncClusterInfo(context.Background(), cluster, true))
+	})
+
+	t.Run("returns promptly when the context is cancelled", func(t *testing.T) {
+		node, cluster := newCluster("127.0.0.1:6399")
+		ctx, cancel := context.WithCancel(context.Background())
+		cancel()
+		require.Error(t, node.SyncClusterInfo(ctx, cluster, true))
+	})
+}
diff --git a/store/cluster_shard.go b/store/cluster_shard.go
index ba1c10d..29c4536 100644
--- a/store/cluster_shard.go
+++ b/store/cluster_shard.go
@@ -24,6 +24,7 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"net"
 	"strings"
 	"sync"
 	"time"
@@ -41,7 +42,6 @@ const (
 	NotMigratingInt = -1
 )
 
-
 // FailoverOptions configures manual failover behavior.
 type FailoverOptions struct {
 	WaitForSync    bool          // whether to wait for replication gap to reach 0
@@ -420,6 +420,12 @@ func (shard *Shard) ToSlotsString() (string, error) {
 	}
 
 	for i, node := range shard.Nodes {
+		// A blank host (":port") would emit a malformed line and register a phantom, unreachable node.
+		// Fail loudly here rather than push corruption to every kvrocks node.
+		host, port, err := net.SplitHostPort(node.Addr())
+		if err != nil || host == "" || port == "" {
+			return "", fmt.Errorf("node %s has invalid address %q", node.ID(), node.Addr())
+		}
 		builder.WriteString(node.ID())
 		builder.WriteByte(' ')
 		builder.WriteString(strings.Replace(node.Addr(), ":", " ", 1))
diff --git a/store/cluster_shard_test.go b/store/cluster_shard_test.go
index 97216fc..7789e9e 100644
--- a/store/cluster_shard_test.go
+++ b/store/cluster_shard_test.go
@@ -99,6 +99,34 @@ func TestToSlotsString_WithFailedSlave(t *testing.T) {
 	require.Contains(t, result, "slave,fail "+master.ID())
 }
 
+func TestToSlotsString_RejectsBlankAddr(t *testing.T) {
+	// An unresolved pod hostname yields a blank host (":6380"); this must not serialize into a
+	// phantom node line — ToSlotsString should fail loudly instead, for either a master or a slave.
+	t.Run("slave", func(t *testing.T) {
+		shard := NewShard()
+		shard.SlotRanges = []SlotRange{{Start: 0, Stop: 100}}
+		master := NewClusterNode("127.0.0.1:6379", "")
+		master.SetRole(RoleMaster)
+		slave := NewClusterNode(":6380", "")
+		slave.SetRole(RoleSlave)
+		shard.Nodes = []Node{master, slave}
+
+		_, err := shard.ToSlotsString()
+		require.Error(t, err)
+	})
+
+	t.Run("master", func(t *testing.T) {
+		shard := NewShard()
+		shard.SlotRanges = []SlotRange{{Start: 0, Stop: 100}}
+		master := NewClusterNode(":6379", "")
+		master.SetRole(RoleMaster)
+		shard.Nodes = []Node{master}
+
+		_, err := shard.ToSlotsString()
+		require.Error(t, err)
+	})
+}
+
 func TestReplicaAppliedReplOffset(t *testing.T) {
 	require.Equal(t, uint64(0), ReplicaAppliedReplOffset(nil))
 	require.Equal(t, uint64(10), ReplicaAppliedReplOffset(&ReplicationInfo{Role: RoleMaster, MasterReplOffset: 10}))
diff --git a/store/cluster_test.go b/store/cluster_test.go
index d9dffdd..576ca6d 100644
--- a/store/cluster_test.go
+++ b/store/cluster_test.go
@@ -22,6 +22,7 @@ package store
 
 import (
 	"context"
+	"errors"
 	"testing"
 
 	"github.com/stretchr/testify/require"
@@ -29,6 +30,31 @@ import (
 	"github.com/apache/kvrocks-controller/consts"
 )
 
+// TestCluster_SyncToNodes asserts SyncToNodes force-pushes the authoritative topology to every node
+// (force=true is what repairs a node drifted at an equal epoch) and surfaces a node-level failure.
+func TestCluster_SyncToNodes(t *testing.T) {
+	ctx := context.Background()
+
+	master := NewClusterMockNode()
+	master.SetRole(RoleMaster)
+	replica := NewClusterMockNode()
+	replica.SetRole(RoleSlave)
+
+	shard := NewShard()
+	shard.SlotRanges = []SlotRange{{Start: 0, Stop: 16383}}
+	shard.Nodes = []Node{master, replica}
+	cluster := &Cluster{Name: "sync", Shards: Shards{shard}}
+	cluster.Version.Store(1)
+
+	require.NoError(t, cluster.SyncToNodes(ctx))
+	require.Equal(t, []bool{true}, master.SyncForceCalls)
+	require.Equal(t, []bool{true}, replica.SyncForceCalls)
+
+	// A node-level push failure is propagated to the caller.
+	replica.SyncErr = errors.New("push rejected")
+	require.Error(t, cluster.SyncToNodes(ctx))
+}
+
 func TestCluster_Clone(t *testing.T) {
 	cluster, err := NewCluster("test", []string{"node1", "node2", "node3"}, 1)
 	require.NoError(t, err)
diff --git a/store/store.go b/store/store.go
index fc319fc..078ec02 100644
--- a/store/store.go
+++ b/store/store.go
@@ -24,9 +24,11 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"net"
+	"sync"
+
 	"github.com/apache/kvrocks-controller/logger"
 	"go.uber.org/zap"
-	"sync"
 
 	"github.com/apache/kvrocks-controller/consts"
 	"github.com/apache/kvrocks-controller/store/engine"
@@ -274,6 +276,12 @@ func (s *ClusterStore) RemoveCluster(ctx context.Context, ns, cluster string) er
 func (s *ClusterStore) CheckNewNodes(ctx context.Context, nodes []string) error {
 	newNodes := make(map[string]bool, 0)
 	for _, node := range nodes {
+		// Reject a blank/half-formed address (e.g. ":6666" from an unresolved hostname) at the API
+		// boundary: it would otherwise serialize into a malformed CLUSTERX SETNODES line and register
+		// a phantom, unreachable node.
+		if host, port, err := net.SplitHostPort(node); err != nil || host == "" || port == "" {
+			return fmt.Errorf("%w: node address must be host:port, got %q", consts.ErrInvalidArgument, node)
+		}
 		newNodes[node] = true
 	}
 
diff --git a/store/store_test.go b/store/store_test.go
index b0517f7..b860943 100644
--- a/store/store_test.go
+++ b/store/store_test.go
@@ -103,5 +103,8 @@ func TestClusterStore(t *testing.T) {
 		require.NoError(t, store.CheckNewNodes(ctx, []string{"127.0.0.1:4444", "127.0.0.1:5555"}))
 		require.NotNil(t, store.CheckNewNodes(ctx, []string{"127.0.0.1:3333", "127.0.0.1:4444"}))
 		require.NotNil(t, store.CheckNewNodes(ctx, []string{"127.0.0.1:2222", "127.0.0.1:3333"}))
+		// Blank / port-less addresses are rejected at the boundary (no phantom nodes).
+		require.Error(t, store.CheckNewNodes(ctx, []string{":6666"}))
+		require.Error(t, store.CheckNewNodes(ctx, []string{"127.0.0.1"}))
 	})
 }