From a5925906c80ef4207ae4400a338bbc1f1a9275e6 Mon Sep 17 00:00:00 2001 From: Lisandro Pin Date: Wed, 21 Oct 2026 15:14:10 +0200 Subject: [PATCH] volume: ask the masters for the leader after losing the heartbeat When the heartbeat stream to the leader failed, currentMaster kept the lost master until a new stream answered, so lookups on a cache miss kept dialing it through the retry delay even with a new leader already elected. On a regular heartbeat error, ask the configured masters for the raft leader and publish it, within the retry delay rather than on top of it. Without an answer the last known leader stays: clearing it would only trade a dead address for an empty one. Co-Authored-By: Claude Opus 5.5 --- weed/server/volume_grpc_client_to_master.go | 22 +++++++++- .../volume_grpc_client_to_master_test.go | 42 +++++++++++++++++++ 2 files changed, 62 insertions(+), 2 deletions(-) diff --git a/weed/server/volume_grpc_client_to_master.go b/weed/server/volume_grpc_client_to_master.go index 2adb379a26..0d00d10823 100644 --- a/weed/server/volume_grpc_client_to_master.go +++ b/weed/server/volume_grpc_client_to_master.go @@ -54,6 +54,19 @@ func (vs *VolumeServer) lookupRaftLeaderMaster(ctx context.Context) (pb.ServerAd return leader, nil } +// refreshLeaderFromPeers publishes the leader the configured masters report, +// so lookups stop going to a master the heartbeat just lost while the loop +// retries. Without an answer within timeout the last known leader stays. +func (vs *VolumeServer) refreshLeaderFromPeers(timeout time.Duration) { + ctx, cancel := context.WithTimeout(context.Background(), timeout) + defer cancel() + leader, err := vs.lookupRaftLeaderMaster(ctx) + if err != nil { + return + } + vs.setCurrentMaster(leader) +} + func isContextDoneErr(err error) bool { if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) { return true @@ -75,7 +88,8 @@ func (vs *VolumeServer) getCurrentMaster() pb.ServerAddress { // setCurrentMaster updates vs.currentMaster under a write lock. The // heartbeat goroutine calls this only with a master known to be the leader: -// one that answered a heartbeat as leader, or the one a follower named. +// one that answered a heartbeat as leader, or the one a follower named, on +// the heartbeat stream or, after the stream failed, in its configuration. func (vs *VolumeServer) setCurrentMaster(master pb.ServerAddress) { vs.currentMasterLock.Lock() vs.currentMaster = master @@ -136,7 +150,11 @@ func (vs *VolumeServer) heartbeat() { } else { // Regular error, reset duplicate retry count duplicateRetryCount = 0 - time.Sleep(vs.pulsePeriod) + // The lookup spends part of the retry delay rather than + // adding to it. + retryAt := time.Now().Add(vs.pulsePeriod) + vs.refreshLeaderFromPeers(vs.pulsePeriod) + time.Sleep(time.Until(retryAt)) } stats.VolumeServerMasterDisconnections.WithLabelValues(master.String()).Inc() diff --git a/weed/server/volume_grpc_client_to_master_test.go b/weed/server/volume_grpc_client_to_master_test.go index 96a6242aa5..1af2744813 100644 --- a/weed/server/volume_grpc_client_to_master_test.go +++ b/weed/server/volume_grpc_client_to_master_test.go @@ -237,3 +237,45 @@ func TestHeartbeatToLeaderPublishesItAndResendsFromSendLoop(t *testing.T) { t.Fatal("heartbeat did not stop") } } + +// After the heartbeat loses its master, lookups move to the leader the other +// masters report instead of dialing the lost one until a new stream answers. +func TestRefreshLeaderFromPeersPublishesReportedLeader(t *testing.T) { + const leader = pb.ServerAddress("leader.example.com:9333.19333") + peer := startFakeMasterServerForLeaderLookup(t, &fakeLeaderConfigServer{leader: string(leader)}) + vs := &VolumeServer{ + SeedMasterNodes: []pb.ServerAddress{peer}, + grpcDialOption: grpc.WithTransportCredentials(insecure.NewCredentials()), + store: &storage.Store{Ip: "127.0.0.1", Port: 8080}, + } + vs.store.MasterFn = vs.getCurrentMaster + vs.setCurrentMaster("lost.example.com:9333.19333") + + vs.refreshLeaderFromPeers(5 * time.Second) + + if got := vs.getCurrentMaster(); got != leader { + t.Fatalf("currentMaster = %q, want reported leader %q", string(got), string(leader)) + } + if got := vs.store.MasterFn(); got != leader { + t.Fatalf("store master = %q, want %q", string(got), string(leader)) + } +} + +// Mid-election no master names a leader; the last known one stays rather +// than an address nobody vouched for. +func TestRefreshLeaderFromPeersKeepsLastLeaderWithoutAnswer(t *testing.T) { + peer := startFakeMasterServerForLeaderLookup(t, &fakeLeaderConfigServer{}) + vs := &VolumeServer{ + SeedMasterNodes: []pb.ServerAddress{peer}, + grpcDialOption: grpc.WithTransportCredentials(insecure.NewCredentials()), + store: &storage.Store{Ip: "127.0.0.1", Port: 8080}, + } + last := pb.ServerAddress("last.example.com:9333.19333") + vs.setCurrentMaster(last) + + vs.refreshLeaderFromPeers(5 * time.Second) + + if got := vs.getCurrentMaster(); got != last { + t.Fatalf("currentMaster = %q, want unchanged %q", string(got), string(last)) + } +}