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)) +}