From b1525a6b7f8efe1d6b0cff47d835461dcbaf88eb Mon Sep 17 00:00:00 2001 From: Fedor Partanskiy Date: Fri, 16 Oct 2026 16:53:25 +0300 Subject: [PATCH] fix: sync a node that stopped hearing its leader A node that falls behind a view change without ever asking to change view has no signal that tells it the cluster moved on. It sits in a view nobody is in any more, and it waits there: it is not the leader, so it has no reason to start a view change, and it keeps receiving the heartbeat of the leader it is stuck with, so it has no reason to time out either. A view change message for the view beyond the next one is what such a node's peers keep sending, but it was recorded and dropped: a view change only accepts messages for the view immediately after the one a node is in, and the view the cluster moved on to is no longer being announced, so there was no other way for the node to find out where the cluster went. A node that hears a quorum of distinct senders (f+1) announce a view beyond the one it is in now syncs, at most once per announced view, so that a faulty sender cannot make it sync over and over again. The announcement carries no signature, so it only triggers the sync: the view itself is still adopted through the authenticated new view message. The per-view sync bookkeeping is reset through a single helper on every path where the view changes, including a restore from the log. The scenario is covered end to end by TestNodeBehindOnIdleViewRejoins in test/view_split_test.go. The scope of it is the window in which the follower still hears its old leader: once the whole cluster is connected, the heartbeat responses and the timeouts rescue both nodes anyway, so this change makes a node recover on the announcement instead of waiting for its timeouts. Based on #700, which fixes the other half of the problem, a node that is one sequence behind the proposal in flight. Signed-off-by: Fedor Partanskiy --- internal/bft/util.go | 13 +++ internal/bft/viewchanger.go | 51 ++++++++- internal/bft/viewchanger_test.go | 10 +- test/view_split_test.go | 181 +++++++++++++++++++++++++++++-- 4 files changed, 239 insertions(+), 16 deletions(-) diff --git a/internal/bft/util.go b/internal/bft/util.go index 4306b00a..13951252 100644 --- a/internal/bft/util.go +++ b/internal/bft/util.go @@ -155,6 +155,19 @@ func (nv *nextViews) sendRecv(next uint64, sender uint64) bool { return next == nv.n[sender] } +// announcedAtOrAbove counts the senders that announced a view at or above the given one. A view +// change message carries no signature, so no single sender can be trusted to be telling the +// truth about it; a quorum of them means at least one of them is honest. +func (nv *nextViews) announcedAtOrAbove(next uint64) uint64 { + var count uint64 + for _, announced := range nv.n { + if announced >= next { + count++ + } + } + return count +} + type incMsg struct { *protos.Message sender uint64 diff --git a/internal/bft/viewchanger.go b/internal/bft/viewchanger.go index 2b1e2539..f2c45830 100644 --- a/internal/bft/viewchanger.go +++ b/internal/bft/viewchanger.go @@ -93,6 +93,10 @@ type ViewChanger struct { backOffFactor uint64 syncedForNewView bool acceptedNewView bool // whether a new view message was already accepted for currView + // syncedForAnnouncedView is the furthest view this node has synced for after hearing a + // peer announce it. It is reset when the node moves to a new view, and it keeps a faulty + // node from making this node sync over and over again. + syncedForAnnouncedView uint64 // Runtime MetricsViewChange *api.MetricsViewChange @@ -145,6 +149,7 @@ func (v *ViewChanger) Start(startViewNumber uint64) { v.MetricsViewChange.CurrentView.Set(float64(v.currView)) v.MetricsViewChange.RealView.Set(float64(v.realView)) v.MetricsViewChange.NextView.Set(float64(v.nextView)) + v.resetSyncedForView() v.lastTick = time.Now() v.lastResend = v.lastTick @@ -283,6 +288,16 @@ func (v *ViewChanger) processMsg(sender uint64, m *protos.Message) { v.processViewChangeMsg(false) return } + if vc.NextView > v.currView+1 { + // The cluster has moved on to a view this node is not in, and is no longer + // announcing it, so there is no other way to find out where the cluster went. + // The announcement carries no signature, so wait for a quorum of distinct senders + // to announce it before acting on it, or one faulty sender could make this node + // sync whenever it liked. Even then the announcement only triggers the sync: the + // view itself is adopted through the authenticated new view message. + v.syncForAnnouncedView(vc.NextView, sender) + return + } if v.nextView == v.currView+1 && // node has already started view change with last view vc.NextView > v.realView && vc.NextView < v.currView+1 && @@ -330,6 +345,33 @@ func (v *ViewChanger) processMsg(sender uint64, m *protos.Message) { } } +// syncForAnnouncedView syncs when a quorum of peers announce a view beyond the one this node is +// in. The view change only accepts messages for the view immediately following the one this +// node is in, and the cluster is no longer announcing the view it moved on to, so without this +// the node waits forever in a view nobody is in. It syncs at most once per announced view, so +// that a faulty node cannot make it sync over and over again. +func (v *ViewChanger) syncForAnnouncedView(announcedView uint64, sender uint64) { + if v.syncedForAnnouncedView >= announcedView { + v.Logger.Debugf("Node %d already synced for a view at or beyond %d, not syncing again", v.SelfID, announcedView) + return + } + if announced := v.nvs.announcedAtOrAbove(announcedView); announced < uint64(v.f)+1 { + v.Logger.Debugf("Node %d got an announcement of view %d from %d, but %d senders announced it and a quorum is %d", v.SelfID, announcedView, sender, announced, uint64(v.f)+1) + return + } + v.Logger.Infof("Node %d is calling sync because %d announced view %d while the current view is %d", v.SelfID, sender, announcedView, v.currView) + v.syncedForAnnouncedView = announcedView + v.Synchronizer.Sync() +} + +// resetSyncedForView clears the record of the views this node has already synced for. It has to +// run wherever the node's view changes, including when it restores one from its log, or a node +// coming back would refuse to sync for the view it is already in the middle of. +func (v *ViewChanger) resetSyncedForView() { + v.syncedForNewView = false + v.syncedForAnnouncedView = v.currView +} + // InformNewView tells the view changer to advance to a new view number func (v *ViewChanger) InformNewView(view uint64) { select { @@ -356,7 +398,7 @@ func (v *ViewChanger) informNewView(view uint64) { v.viewDataMsgs.clear(v.N) v.checkTimeout = false v.backOffFactor = 1 // reset - v.syncedForNewView = false + v.resetSyncedForView() v.acceptedNewView = true // the node is already in this view, so a new view message for it is not needed v.RequestsTimer.RestartTimers() } @@ -428,9 +470,9 @@ func (v *ViewChanger) processViewChangeMsg(restore bool) { v.currView = v.nextView v.MetricsViewChange.CurrentView.Set(float64(v.currView)) v.viewChangeMsgs.clear(v.N) - v.viewDataMsgs.clear(v.N) // clear because currView changed - v.syncedForNewView = false // clear because currView changed - v.acceptedNewView = false // clear because currView changed + v.viewDataMsgs.clear(v.N) // clear because currView changed + v.resetSyncedForView() // clear because currView changed + v.acceptedNewView = false // clear because currView changed msg := v.prepareViewDataMsg() leader := v.getLeader() if leader == v.SelfID { @@ -1203,6 +1245,7 @@ func (v *ViewChanger) processNewViewMsg(msg *protos.NewView) { v.realView = v.currView v.MetricsViewChange.RealView.Set(float64(v.realView)) v.nvs.clear() + v.resetSyncedForView() v.Controller.ViewChanged(v.currView, mySequence+1) v.RequestsTimer.RestartTimers() diff --git a/internal/bft/viewchanger_test.go b/internal/bft/viewchanger_test.go index 0918d43f..dfba19c6 100644 --- a/internal/bft/viewchanger_test.go +++ b/internal/bft/viewchanger_test.go @@ -165,6 +165,8 @@ func TestViewChangeProcess(t *testing.T) { controller.On("AbortView", mock.Anything) state := &mocks.State{} state.On("Save", mock.Anything).Return(nil) + synchronizer := &mocks.Synchronizer{} + synchronizer.On("Sync") vc := &bft.ViewChanger{ SelfID: 0, @@ -174,6 +176,7 @@ func TestViewChangeProcess(t *testing.T) { Signer: signer, Logger: log, RequestsTimer: reqTimer, + Synchronizer: synchronizer, Ticker: make(chan time.Time), InFlight: &bft.InFlightData{}, Checkpoint: &types.Checkpoint{}, @@ -204,7 +207,9 @@ func TestViewChangeProcess(t *testing.T) { vc.HandleMessage(3, viewChangeMsg) } - // sending viewChange messages with bigger view doesn't make a difference + // a quorum of senders announcing a view beyond the next one does make a difference: + // the cluster moved on and is no longer announcing the view this node is in, so the + // node syncs to find out where it went msg3 := proto.Clone(viewChangeMsg).(*protos.Message) msg3.GetViewChange().NextView = 3 vc.HandleMessage(2, msg3) @@ -227,6 +232,9 @@ func TestViewChangeProcess(t *testing.T) { assert.NotNil(t, m.GetViewData()) comm.AssertCalled(t, "SendConsensus", uint64(2), mock.Anything) + // asserted after the reads above, which is what waits for the view changer to + // have processed the messages that were just sent to it + synchronizer.AssertNumberOfCalls(t, "Sync", 1) reqTimer.AssertNumberOfCalls(t, "StopTimers", 2) controller.AssertNumberOfCalls(t, "AbortView", 4) state.AssertNumberOfCalls(t, "Save", 2) diff --git a/test/view_split_test.go b/test/view_split_test.go index af1385b2..9be0b823 100644 --- a/test/view_split_test.go +++ b/test/view_split_test.go @@ -30,7 +30,12 @@ type viewTracker struct { quorums map[uint64]int viewData map[uint64]int syncs map[uint64]int - views map[uint64]uint64 + // announcedSyncs counts only the syncs a node made because peers announced a view beyond + // the one it is in. The generic syncs counter cannot tell that path apart from the ones + // the view change timeout and the heartbeat already made before the fix, so a test that + // asserts on it proves nothing about the announcement path. + announcedSyncs map[uint64]int + views map[uint64]uint64 // proposals is the furthest sequence each node processed a proposal at. The leader broadcasts // the pre-prepare while processing the proposal, and it is broadcast only once, so this is // what tells a test whether a node that joins a view later has already missed it. @@ -39,12 +44,13 @@ type viewTracker struct { func newViewTracker() *viewTracker { return &viewTracker{ - attempts: make(map[uint64]int), - quorums: make(map[uint64]int), - viewData: make(map[uint64]int), - syncs: make(map[uint64]int), - views: make(map[uint64]uint64), - proposals: make(map[uint64]uint64), + attempts: make(map[uint64]int), + quorums: make(map[uint64]int), + viewData: make(map[uint64]int), + syncs: make(map[uint64]int), + announcedSyncs: make(map[uint64]int), + views: make(map[uint64]uint64), + proposals: make(map[uint64]uint64), } } @@ -76,10 +82,13 @@ func (vt *viewTracker) hook(id uint64) func(zapcore.Entry) error { vt.bump(vt.viewData, id) case strings.Contains(msg, "got a quorum of viewData messages"): vt.bump(vt.quorums, id) + case strings.Contains(msg, "is calling sync because") && strings.Contains(msg, "announced view"): + vt.bump(vt.announcedSyncs, id) + vt.bump(vt.syncs, id) case strings.Contains(msg, "get msg from syncChan"): // the controller logs this once for every sync it performs, whichever part of the - // library asked for it: the view change timeout or a heartbeat saying the cluster - // is ahead of it + // library asked for it: the view change timeout, an announcement of a view beyond + // the one this node is in, or a heartbeat saying the cluster is ahead of it vt.bump(vt.syncs, id) } return nil @@ -147,6 +156,14 @@ func (vt *viewTracker) syncsOf(id uint64) int { return vt.syncs[id] } +// announcedSyncsOf returns how many times the node synchronized because a quorum of peers had +// announced a view beyond the one it was in. +func (vt *viewTracker) announcedSyncsOf(id uint64) int { + vt.mu.Lock() + defer vt.mu.Unlock() + return vt.announcedSyncs[id] +} + func (vt *viewTracker) proposalOf(id uint64) uint64 { vt.mu.Lock() defer vt.mu.Unlock() @@ -157,8 +174,8 @@ func (vt *viewTracker) proposalOf(id uint64) uint64 { func (vt *viewTracker) describe(ids []uint64) string { var b strings.Builder for _, id := range ids { - fmt.Fprintf(&b, "\n node %d: view=%d, view change attempts=%d, view data sent=%d, view data quorums=%d, syncs=%d, last proposal=%d", - id, vt.viewOf(id), vt.attemptsOf(id), vt.viewDataOf(id), vt.quorumsOf(id), vt.syncsOf(id), vt.proposalOf(id)) + fmt.Fprintf(&b, "\n node %d: view=%d, view change attempts=%d, view data sent=%d, view data quorums=%d, syncs=%d (announced view: %d), last proposal=%d", + id, vt.viewOf(id), vt.attemptsOf(id), vt.viewDataOf(id), vt.quorumsOf(id), vt.syncsOf(id), vt.announcedSyncsOf(id), vt.proposalOf(id)) } return b.String() } @@ -228,6 +245,13 @@ func decideFirstRequest(t *testing.T, nodes []*App) { } } +// cutAll breaks the link between two nodes in both directions. DisconnectFrom only closes the +// sending side, so both sides have to be cut for the split to be a real one. +func cutAll(nodes []*App, a, b uint64) { + nodes[a-1].DisconnectFrom(b) + nodes[b-1].DisconnectFrom(a) +} + // heal restores every link of the cluster. func heal(nodes []*App, ids []uint64) { for _, id := range ids { @@ -316,6 +340,29 @@ func viewsReached(t *testing.T, vt *viewTracker, ids []uint64, view uint64, time }, timeout, 100*time.Millisecond, msg) } +// attemptsReached waits until every listed node has tried to change view at least n times. +func attemptsReached(t *testing.T, vt *viewTracker, ids []uint64, n int, timeout time.Duration, msg string) bool { + t.Helper() + return assert.Eventually(t, func() bool { + for _, id := range ids { + if vt.attemptsOf(id) < n { + return false + } + } + return true + }, timeout, 100*time.Millisecond, msg) +} + +// announcedSyncReached waits until the node synchronized because a quorum of peers announced a +// view beyond the one it was in. The generic sync counter cannot be used for this: the view +// change timeout and the heartbeat already synchronized a node before the fix as well. +func announcedSyncReached(t *testing.T, vt *viewTracker, id uint64, timeout time.Duration, msg string) bool { + t.Helper() + return assert.Eventually(t, func() bool { + return vt.announcedSyncsOf(id) > 0 + }, timeout, 100*time.Millisecond, msg) +} + // proposalProcessed waits until the node processed a proposal at the given sequence. The leader // broadcasts the pre-prepare while processing the proposal and broadcasts it only once, so a node // that joins the view after this point cannot receive it any more. @@ -443,3 +490,115 @@ func TestNodeOneSequenceBehindRejoins(t *testing.T) { t.Logf("state after the cluster recovered:%s", vt.describe(ids)) } + +// TestNodeBehindOnIdleViewRejoins covers the node that fell behind without ever having asked to +// change view. +// +// This is the case the description of the fix is about: the node sits in a view nobody is in any +// more, and nothing on its own moves it. Node 1 leads view 0 and keeps sending its heartbeat to +// node 2, so node 2 never times out and never starts a view change. The other five move on to +// view 2, and then split among themselves: the three in the middle are cut off from the leader +// of view 2 and keep announcing view 3, which they cannot install with only three nodes. +// +// Node 2 is let back in to the three in the middle, while node 1 stays cut off. Node 1 cannot +// gather the quorum of heartbeat responses that would synchronize it, and node 2 still hears its +// leader, so the only signal left to either of them is the announcements: a quorum of the peers +// announce a view beyond the one this node is in. That is what makes node 2 synchronize and find +// out where the cluster went. Before the fix a view change message for a view beyond the next one +// was recorded and dropped, so node 2 waited forever in a view nobody was in. +// +// Note the scope of this scenario: the announcements are what node 2 needs while it still hears +// the leader it is stuck with. Once the whole cluster is connected, node 1 gathers the heartbeat +// responses that synchronize it, and node 2 follows it out after its heartbeat timeout, so the +// timeouts do eventually rescue both. This test therefore pins down that the announcement path is +// taken and that it is what the node acts on, rather than that it is the only way back. +func TestNodeBehindOnIdleViewRejoins(t *testing.T) { + t.Parallel() + + const numberOfNodes = 7 + + nodes, ids, vt := newSplitCluster(t, numberOfNodes) + decideFirstRequest(t, nodes) + + // the node that ends up behind, and the leader it is stuck with + behind := []uint64{1, 2} + // the nodes that end up furthest ahead + ahead := []uint64{3, 7} + // the nodes that end up in the middle, stuck one vote short of changing view + middle := []uint64{4, 5, 6} + + done := make(chan struct{}) + defer close(done) + var counter uint64 + accelerateTime(nodes, done, true, true, &counter) + + // The two behind are cut off together, but they keep hearing each other: node 1 is the + // leader of view 0 and keeps sending heartbeats to node 2, so neither of them starts a view + // change, even though their timers are running. + for _, id := range behind { + for _, other := range append(append([]uint64{}, ahead...), middle...) { + cutAll(nodes, id, other) + } + } + + // With the leader of view 1 among the two that are cut off, the other five install view 2. + if !viewsReached(t, vt, append(append([]uint64{}, ahead...), middle...), 2, 90*time.Second, + "the five nodes that moved on did not start view 2") { + t.Fatalf("the five nodes that moved on did not start view 2:%s", vt.describe(ids)) + } + // the two behind did not move, and neither did the leader of the view they are stuck with + if v := vt.viewOf(1); v != 0 { + t.Fatalf("node 1 changed view while it was cut off: view=%d:%s", v, vt.describe(ids)) + } + if v := vt.viewOf(2); v != 0 { + t.Fatalf("node 2 changed view while it was cut off: view=%d:%s", v, vt.describe(ids)) + } + + // The five now split. The three in the middle keep their view's leader cut off, so they + // cannot gather the votes they need to change view on their own, and keep announcing view 3. + // The two in front stay in view 2, which is the view the cluster is actually in. + for _, id := range middle { + for _, other := range ahead { + cutAll(nodes, id, other) + } + } + if !attemptsReached(t, vt, middle, 3, 90*time.Second, + "the three nodes in the middle did not start announcing a new view") { + t.Fatalf("the three nodes in the middle did not start announcing a new view:%s", vt.describe(ids)) + } + for _, id := range append(append([]uint64{}, ahead...), middle...) { + if v := vt.viewOf(id); v != 2 { + t.Fatalf("node %d is not in view 2 after the split: view=%d:%s", id, v, vt.describe(ids)) + } + } + + t.Logf("state with the cluster in view 2 and the middle three announcing view 3:%s", vt.describe(ids)) + + // Let node 2 back in to the three in the middle. Node 1 stays cut off: it would gather the + // f+1 heartbeat responses that synchronize it as soon as it could reach the cluster, and + // would then pull node 2 along, so the announcement path would never be reached. + for _, id := range middle { + nodes[id-1].ConnectTo(behind[1]) + } + nodes[behind[1]-1].ConnectTo(middle[0]) + nodes[behind[1]-1].ConnectTo(middle[1]) + nodes[behind[1]-1].ConnectTo(middle[2]) + + // This is the assertion the fix exists for: node 2 synchronized because a quorum of its + // peers announced a view it is not in. + if !announcedSyncReached(t, vt, behind[1], 60*time.Second, + "node 2 never synchronized because of an announced view, so it never had a chance to learn where the cluster went") { + t.Fatalf("node 2 never synchronized because of an announced view:%s", vt.describe(ids)) + } + if !viewsReached(t, vt, behind[1:], 2, 60*time.Second, "node 2 did not reach the view of the cluster") { + t.Fatalf("node 2 did not reach the view of the cluster:%s", vt.describe(ids)) + } + + // the rest of the cluster is connected again and has to decide on its own + heal(nodes, ids) + + nodes[0].Submit(Request{ID: "2", ClientID: "alice"}) + deliverRequest(t, nodes, ids, 2, vt) + + t.Logf("state after the cluster recovered:%s", vt.describe(ids)) +}