fix(wdclient,volume): compare master leader with ServerAddress.Equals (#9089)

* fix(wdclient,volume): compare master leader with ServerAddress.Equals

Raft leader is advertised as host:httpPort.grpcPort, but clients dial
host:httpPort. Raw string comparison against VolumeLocation.Leader /
HeartbeatResponse.Leader therefore never matches, causing the
masterclient and the volume server heartbeat loop to continuously
"redirect" to the already-connected master, tearing down the stream
and reconnecting.

Use ServerAddress.Equals, which normalizes the grpc-port suffix.

* fix(filer,mq): compare ServerAddress via Equals in two more sites

filer bootstrap skip (MaybeBootstrapFromOnePeer) and the broker's local
partition assignment check both compared a wire-supplied address string
against the local self ServerAddress with raw string equality. Both are
vulnerable to the same plain-vs-host:port.grpcPort mismatch as the
masterclient/volume heartbeat sites: filer would bootstrap from itself,
and the broker would fail to claim a partition it was actually assigned.
Route both through ServerAddress.Equals.

* fix(master,shell): more ServerAddress comparisons via Equals

- raft_server_handlers.go HealthzHandler: s.serverAddr == leader would
  skip the child-lock check on the real leader when the two carry
  different plain/grpc-suffix forms, returning 200 OK instead of 423.
- master_server.go SetRaftServer leader-change callback: the
  Leader() == Name() guard for ensureTopologyId could disagree with
  topology.IsLeader() (which already uses Equals), so leader-only
  initialization could be skipped after an election.
- command_volume_merge.go isReplicaServer: the -target guard compared
  user-supplied host:port against NewServerAddressFromDataNode(...) with
  ==, letting an existing replica slip through when topology carries
  the embedded gRPC port.

All routed through pb.ServerAddress.Equals.

* fix(mq,cluster): more ServerAddress comparisons via Equals

- broker_grpc_lookup.go GetTopicPublishers/GetTopicSubscribers: the
  partition ownership check gated listing on raw
  LeaderBroker == BrokerAddress().String(), so listings silently omitted
  partitions hosted locally when the assignment carried the other
  host:port / host:port.grpcPort form.
- lock_client.go: LockHostMovedTo comparison and the seedFiler fallback
  guard both used raw string equality against configured filer
  addresses (which may be plain host:port while LockHostMovedTo comes
  back suffixed), causing spurious host-change churn and blocking the
  seed-filer fallback.

* fix(mq): more ServerAddress comparisons via Equals

- pub_balancer/allocate.go EnsureAssignmentsToActiveBrokers: direct
  activeBrokers.Get() lookup missed brokers when a persisted assignment
  carried a different address encoding than the registered broker key,
  triggering a bogus reassignment on every read/write cycle. Added a
  findActiveBroker helper that falls back to an Equals-based scan and
  canonicalizes the assignment in place so later writes are stable.
- broker_grpc_lookup.go isLockOwner: used raw string equality between
  LockOwner() and BrokerAddress().String(), so a lock owner could fail
  to recognize itself and proxy local lookup/config/admin RPCs away.
- pub_client/scheduler.go onEachAssignments: reused publisher jobs only
  on exact LeaderBroker match, so an encoding flip in lookup results
  tore down and recreated a stream to the same broker.

weed/cluster/lock_client.go

@@ -267,7 +267,7 @@ func (lock *LiveLock) doLock(lockDuration time.Duration) (errorMessage string, e
 		}
 		if resp != nil {
 			errorMessage = resp.Error
-			if resp.LockHostMovedTo != "" && resp.LockHostMovedTo != string(previousHostFiler) {
+			if resp.LockHostMovedTo != "" && !pb.ServerAddress(resp.LockHostMovedTo).Equals(previousHostFiler) {
 				// Only log if the host actually changed
 				glog.V(2).Infof("LOCK: Host changed from %s to %s for key=%s", previousHostFiler, resp.LockHostMovedTo, lock.key)
 				lock.hostFiler = pb.ServerAddress(resp.LockHostMovedTo)
@@ -289,7 +289,7 @@ func (lock *LiveLock) doLock(lockDuration time.Duration) (errorMessage string, e
 		return err
 	})
 
-	if err != nil && lock.hostFiler != lock.lc.seedFiler {
+	if err != nil && !lock.hostFiler.Equals(lock.lc.seedFiler) {
 		lock.consecutiveFailures++
 		// Fall back to seed filer after 3 consecutive connection failures
 		if lock.consecutiveFailures >= 3 {

weed/filer/filer.go

@@ -108,7 +108,7 @@ func (f *Filer) MaybeBootstrapFromOnePeer(self pb.ServerAddress, existingNodes [
 		return existingNodes[i].CreatedAtNs < existingNodes[j].CreatedAtNs
 	})
 	earliestNode := existingNodes[0]
-	if earliestNode.Address == string(self) {
+	if pb.ServerAddress(earliestNode.Address).Equals(self) {
 		return
 	}
 

weed/mq/broker/broker_grpc_lookup.go

@@ -336,7 +336,7 @@ func (b *MessageQueueBroker) GetTopicPublishers(ctx context.Context, request *mq
 	// Collect publishers from each partition that is hosted on this broker
 	for _, assignment := range conf.BrokerPartitionAssignments {
 		// Only collect from partitions where this broker is the leader
-		if assignment.LeaderBroker == b.option.BrokerAddress().String() {
+		if pb.ServerAddress(assignment.LeaderBroker).Equals(b.option.BrokerAddress()) {
 			partition := topic.FromPbPartition(assignment.Partition)
 			if localPartition := b.localTopicManager.GetLocalPartition(t, partition); localPartition != nil {
 				// Get publisher information from local partition
@@ -390,7 +390,7 @@ func (b *MessageQueueBroker) GetTopicSubscribers(ctx context.Context, request *m
 	// Collect subscribers from each partition that is hosted on this broker
 	for _, assignment := range conf.BrokerPartitionAssignments {
 		// Only collect from partitions where this broker is the leader
-		if assignment.LeaderBroker == b.option.BrokerAddress().String() {
+		if pb.ServerAddress(assignment.LeaderBroker).Equals(b.option.BrokerAddress()) {
 			partition := topic.FromPbPartition(assignment.Partition)
 			if localPartition := b.localTopicManager.GetLocalPartition(t, partition); localPartition != nil {
 				// Get subscriber information from local partition
@@ -430,5 +430,5 @@ func (b *MessageQueueBroker) GetTopicSubscribers(ctx context.Context, request *m
 }
 
 func (b *MessageQueueBroker) isLockOwner() bool {
-	return b.lockAsBalancer.LockOwner() == b.option.BrokerAddress().String()
+	return pb.ServerAddress(b.lockAsBalancer.LockOwner()).Equals(b.option.BrokerAddress())
 }

weed/mq/broker/broker_topic_conf_read_write.go

@@ -13,6 +13,7 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/mq/logstore"
 	"github.com/seaweedfs/seaweedfs/weed/mq/pub_balancer"
 	"github.com/seaweedfs/seaweedfs/weed/mq/topic"
+	"github.com/seaweedfs/seaweedfs/weed/pb"
 	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
 	"github.com/seaweedfs/seaweedfs/weed/pb/mq_pb"
 )
@@ -147,12 +148,12 @@ func (b *MessageQueueBroker) genLocalPartitionFromFiler(t topic.Topic, partition
 	for _, assignment := range conf.BrokerPartitionAssignments {
 		assignmentPartition := topic.FromPbPartition(assignment.Partition)
 		glog.V(4).Infof("checking assignment: LeaderBroker=%s, Partition=%s", assignment.LeaderBroker, assignmentPartition)
-		glog.V(4).Infof("comparing self=%s with LeaderBroker=%s: %v", self, assignment.LeaderBroker, assignment.LeaderBroker == string(self))
+		glog.V(4).Infof("comparing self=%s with LeaderBroker=%s: %v", self, assignment.LeaderBroker, pb.ServerAddress(assignment.LeaderBroker).Equals(self))
 		glog.V(4).Infof("comparing partition=%s with assignmentPartition=%s: %v", partition.String(), assignmentPartition.String(), partition.Equals(assignmentPartition))
 		glog.V(4).Infof("logical comparison (RangeStart, RangeStop only): %v", partition.LogicalEquals(assignmentPartition))
 		glog.V(4).Infof("partition details: RangeStart=%d, RangeStop=%d, RingSize=%d, UnixTimeNs=%d", partition.RangeStart, partition.RangeStop, partition.RingSize, partition.UnixTimeNs)
 		glog.V(4).Infof("assignmentPartition details: RangeStart=%d, RangeStop=%d, RingSize=%d, UnixTimeNs=%d", assignmentPartition.RangeStart, assignmentPartition.RangeStop, assignmentPartition.RingSize, assignmentPartition.UnixTimeNs)
-		if assignment.LeaderBroker == string(self) && partition.LogicalEquals(assignmentPartition) {
+		if pb.ServerAddress(assignment.LeaderBroker).Equals(self) && partition.LogicalEquals(assignmentPartition) {
 			glog.V(4).Infof("Creating local partition for %s %s", t, partition)
 			localPartition = topic.NewLocalPartition(partition, b.option.LogFlushInterval, b.genLogFlushFunc(t, partition), logstore.GenMergedReadFunc(b, t, partition))
 

weed/mq/client/pub_client/scheduler.go

@@ -91,7 +91,7 @@ func (p *TopicPublisher) onEachAssignments(generation int, assignments []*mq_pb.
 		if hasExistingJob {
 			var existingJob *EachPartitionPublishJob
 			existingJob = p.jobs[i]
-			if existingJob.BrokerPartitionAssignment.LeaderBroker == assignment.LeaderBroker {
+			if pb.ServerAddress(existingJob.BrokerPartitionAssignment.LeaderBroker).Equals(pb.ServerAddress(assignment.LeaderBroker)) {
 				existingJob.generation = generation
 				jobs = append(jobs, existingJob)
 				continue

weed/mq/pub_balancer/allocate.go

@@ -6,10 +6,30 @@ import (
 
 	cmap "github.com/orcaman/concurrent-map/v2"
 	"github.com/seaweedfs/seaweedfs/weed/glog"
+	"github.com/seaweedfs/seaweedfs/weed/pb"
 	"github.com/seaweedfs/seaweedfs/weed/pb/mq_pb"
 	"github.com/seaweedfs/seaweedfs/weed/pb/schema_pb"
 )
 
+// findActiveBroker looks up a broker in activeBrokers, tolerating address
+// encoding differences (plain host:port vs host:port.grpcPort) by falling
+// back to an Equals-based scan when the direct key lookup misses.
+func findActiveBroker(activeBrokers cmap.ConcurrentMap[string, *BrokerStats], addr string) (string, bool) {
+	if addr == "" {
+		return "", false
+	}
+	if _, found := activeBrokers.Get(addr); found {
+		return addr, true
+	}
+	target := pb.ServerAddress(addr)
+	for item := range activeBrokers.IterBuffered() {
+		if pb.ServerAddress(item.Key).Equals(target) {
+			return item.Key, true
+		}
+	}
+	return "", false
+}
+
 func AllocateTopicPartitions(brokers cmap.ConcurrentMap[string, *BrokerStats], partitionCount int32) (assignments []*mq_pb.BrokerPartitionAssignment) {
 	// divide the ring into partitions
 	now := time.Now().UnixNano()
@@ -91,15 +111,21 @@ func EnsureAssignmentsToActiveBrokers(activeBrokers cmap.ConcurrentMap[string, *
 		count := 0
 		if assignment.LeaderBroker == "" {
 			count++
-		} else if _, found := activeBrokers.Get(assignment.LeaderBroker); !found {
+		} else if canonical, found := findActiveBroker(activeBrokers, assignment.LeaderBroker); !found {
 			assignment.LeaderBroker = ""
 			count++
+		} else if canonical != assignment.LeaderBroker {
+			assignment.LeaderBroker = canonical
+			hasChanges = true
 		}
 		if assignment.FollowerBroker == "" {
 			count++
-		} else if _, found := activeBrokers.Get(assignment.FollowerBroker); !found {
+		} else if canonical, found := findActiveBroker(activeBrokers, assignment.FollowerBroker); !found {
 			assignment.FollowerBroker = ""
 			count++
+		} else if canonical != assignment.FollowerBroker {
+			assignment.FollowerBroker = canonical
+			hasChanges = true
 		}
 
 		if count > 0 {

weed/server/master_server.go

@@ -220,7 +220,7 @@ func (ms *MasterServer) SetRaftServer(raftServer *RaftServer) {
 			if ms.Topo.RaftServer.Leader() != "" {
 				glog.V(0).Infof("[%s] %s becomes leader.", ms.Topo.RaftServer.Name(), ms.Topo.RaftServer.Leader())
 				ms.Topo.SetLastLeaderChangeTime(time.Now())
-				if ms.Topo.RaftServer.Leader() == ms.Topo.RaftServer.Name() {
+				if pb.ServerAddress(ms.Topo.RaftServer.Leader()).Equals(pb.ServerAddress(ms.Topo.RaftServer.Name())) {
 					go ms.ensureTopologyId()
 				}
 			}

weed/server/raft_server_handlers.go

@@ -36,7 +36,7 @@ func (s *RaftServer) HealthzHandler(w http.ResponseWriter, r *http.Request) {
 		w.WriteHeader(http.StatusServiceUnavailable)
 		return
 	}
-	if s.serverAddr == leader {
+	if s.serverAddr.Equals(leader) {
 		expBackoff := backoff.NewExponentialBackOff()
 		expBackoff.InitialInterval = 20 * time.Millisecond
 		expBackoff.MaxInterval = 1 * time.Second

weed/server/volume_grpc_client_to_master.go

@@ -179,7 +179,7 @@ func (vs *VolumeServer) doHeartbeatWithRetry(masterAddress pb.ServerAddress, grp
 					}
 				}
 			}
-			if in.GetLeader() != "" && string(vs.currentMaster) != in.GetLeader() {
+			if in.GetLeader() != "" && !vs.currentMaster.Equals(pb.ServerAddress(in.GetLeader())) {
 				glog.V(0).Infof("Volume Server found a new master newLeader: %v instead of %v", in.GetLeader(), vs.currentMaster)
 				newLeader = pb.ServerAddress(in.GetLeader())
 				doneChan <- nil

weed/shell/command_volume_merge.go

@@ -493,7 +493,7 @@ func ensureVolumeReadonly(commandEnv *CommandEnv, replicas []*VolumeReplica) ([]
 
 func isReplicaServer(target pb.ServerAddress, replicas []*VolumeReplica) bool {
 	for _, replica := range replicas {
-		if pb.NewServerAddressFromDataNode(replica.location.dataNode) == target {
+		if pb.NewServerAddressFromDataNode(replica.location.dataNode).Equals(target) {
 			return true
 		}
 	}

weed/wdclient/masterclient.go

@@ -264,7 +264,7 @@ func (mc *MasterClient) tryConnectToMaster(ctx context.Context, master pb.Server
 
 		// check if it is the leader to determine whether to reset the vidMap
 		if resp.VolumeLocation != nil {
-			if resp.VolumeLocation.Leader != "" && string(master) != resp.VolumeLocation.Leader {
+			if resp.VolumeLocation.Leader != "" && !master.Equals(pb.ServerAddress(resp.VolumeLocation.Leader)) {
 				glog.V(1).Infof("master %v redirected to leader %v", master, resp.VolumeLocation.Leader)
 				nextHintedLeader = pb.ServerAddress(resp.VolumeLocation.Leader)
 				stats.MasterClientConnectCounter.WithLabelValues(stats.RedirectedToLeader).Inc()
@@ -295,8 +295,8 @@ func (mc *MasterClient) tryConnectToMaster(ctx context.Context, master pb.Server
 			if resp.VolumeLocation != nil {
 				// Check for leader change during the stream
 				// If master announces a new leader, reconnect to it
-				if resp.VolumeLocation.Leader != "" && string(mc.GetMaster(ctx)) != resp.VolumeLocation.Leader {
-					glog.V(1).Infof("currentMaster %v redirected to leader %v", mc.GetMaster(ctx), resp.VolumeLocation.Leader)
+				if currentMaster := mc.GetMaster(ctx); resp.VolumeLocation.Leader != "" && !currentMaster.Equals(pb.ServerAddress(resp.VolumeLocation.Leader)) {
+					glog.V(1).Infof("currentMaster %v redirected to leader %v", currentMaster, resp.VolumeLocation.Leader)
 					nextHintedLeader = pb.ServerAddress(resp.VolumeLocation.Leader)
 					stats.MasterClientConnectCounter.WithLabelValues(stats.RedirectedToLeader).Inc()
 					return nil