fix: sync a node that stopped hearing its leader

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

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()

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)

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