Fix lagging node view change deadlock

internal/bft/controller.go

@@ -659,7 +659,12 @@ func (c *Controller) sync() (viewNum uint64, seq uint64, decisions uint64) {
 		if response.View <= controllerViewNum && latestDecisionViewNum < controllerViewNum {
 			return 0, 0, 0 // no new view to report
 		}
-		if response.View > newViewNum && response.Seq == latestDecisionSeq+1 {
+		// The peers report the sequence they are at. Accept their view as long as it is
+		// ahead of what this node knows, no matter how many sequences ahead they are: the
+		// synchronizer catches up the decisions, and a node that refuses to move on because
+		// the cluster is more than one sequence ahead would stay in a view the cluster left
+		// and no quorum could be formed again.
+		if response.View > newViewNum && response.Seq > latestDecisionSeq {
 			c.Logger.Infof("Node %d collected state with view %d and sequence %d", c.ID, response.View, response.Seq)
 			newViewToSave := &protos.SavedMessage{
 				Content: &protos.SavedMessage_NewView{

internal/bft/controller_sync_test.go

@@ -30,6 +30,49 @@ func (s *stubComm) SendTransaction(targetID uint64, request []byte)  {}
 func (s *stubComm) Nodes() []uint64                                  { return []uint64{1, 2, 3, 4} }
 func (s *stubComm) BroadcastConsensus(m *protos.Message)             {}
 
+// stateRecorder records the last message saved to the state.
+type stateRecorder struct {
+	saved *protos.SavedMessage
+}
+
+func (s *stateRecorder) Save(m *protos.SavedMessage) error {
+	s.saved = m
+	return nil
+}
+
+func (s *stateRecorder) Restore(v *View) error { return nil }
+
+// stubRequestsTimer is a no-op implementation of RequestsTimer.
+type stubRequestsTimer struct{}
+
+func (s *stubRequestsTimer) StopTimers()                             {}
+func (s *stubRequestsTimer) RestartTimers()                          {}
+func (s *stubRequestsTimer) RemoveRequest(r types.RequestInfo) error { return nil }
+
+// peerStateComm answers state transfer requests with a fixed view and sequence, the way the
+// other nodes of the cluster do.
+type peerStateComm struct {
+	stubComm
+	view       uint64
+	seq        uint64
+	onResponse func(sender uint64, m *protos.Message)
+}
+
+func (c *peerStateComm) SendConsensus(targetID uint64, m *protos.Message) {
+	if m.GetStateTransferRequest() == nil || c.onResponse == nil {
+		return
+	}
+	// the peers answer asynchronously, they are not inline with the request
+	go c.onResponse(targetID, &protos.Message{
+		Content: &protos.Message_StateTransferResponse{
+			StateTransferResponse: &protos.StateTransferResponse{
+				ViewNum:  c.view,
+				Sequence: c.seq,
+			},
+		},
+	})
+}
+
 // TestSyncDecisionsInView tests that the sync() method returns the correct
 // DecisionsInView value in two scenarios:
 //   - same height: sync returns the same sequence as the controller, so
@@ -148,3 +191,117 @@ func TestSyncDecisionsInView(t *testing.T) {
 			"DecisionsInView should be latestDecisionDecisions+1 when sync returns higher height")
 	})
 }
+
+// the sequence the lagging node knows about, and the view it is stuck in
+const (
+	controllerSeq       = uint64(100)
+	controllerView      = uint64(67)
+	controllerDecisions = uint64(3)
+	clusterView         = uint64(69)
+)
+
+// TestSyncLearnsViewFromPeers covers the case where the cluster moved to a higher view without
+// making a decision, so the latest decision says nothing about that view. The only source of
+// the current view is then the state the other nodes report, and the node has to be able to
+// move to it, otherwise it stays in a view the rest of the cluster left and no quorum can be
+// formed for transactions again.
+func TestSyncLearnsViewFromPeers(t *testing.T) {
+	// the peers report the sequence they are at. It is higher than the one the lagging node
+	// knows, which is what happens when the cluster also decided transactions while the node
+	// was stuck, and it is what the node has to catch up with before it can rejoin.
+	for _, test := range []struct {
+		description string
+		peerSeq     uint64
+	}{
+		{description: "the cluster did not decide anything while the node was stuck", peerSeq: controllerSeq + 1},
+		{description: "the cluster decided transactions while the node was stuck", peerSeq: controllerSeq + 10},
+	} {
+		t.Run(test.description, func(t *testing.T) {
+			runSyncAndAssertViewLearned(t, test.peerSeq)
+		})
+	}
+}
+
+func runSyncAndAssertViewLearned(t *testing.T, peerSeq uint64) {
+	t.Helper()
+
+	basicLog, err := zap.NewDevelopment()
+	assert.NoError(t, err)
+	log := basicLog.Sugar()
+
+	// the node is in view 67 and has not decided anything since, so its checkpoint knows
+	// nothing about the view the cluster moved to
+	checkpoint := &types.Checkpoint{}
+	checkpoint.Set(types.Proposal{
+		Metadata: MarshalOrPanic(&protos.ViewMetadata{
+			LatestSequence:  controllerSeq,
+			ViewId:          controllerView,
+			DecisionsInView: controllerDecisions,
+		}),
+	}, nil)
+
+	// the synchronizer returns the same decision the node already has, since the cluster made
+	// no progress on transactions the node can see while the view was changing
+	syncResponse := types.SyncResponse{
+		Latest: types.Decision{
+			Proposal: types.Proposal{
+				Metadata: MarshalOrPanic(&protos.ViewMetadata{
+					LatestSequence:  controllerSeq,
+					ViewId:          controllerView,
+					DecisionsInView: controllerDecisions,
+				}),
+			},
+		},
+	}
+
+	// the peers report that the cluster is now in view 69
+	collector := &StateCollector{
+		SelfID:         1,
+		N:              4,
+		Logger:         log,
+		CollectTimeout: time.Second,
+	}
+	collector.Start()
+	t.Cleanup(collector.Stop)
+
+	comm := &peerStateComm{view: clusterView, seq: peerSeq}
+	comm.onResponse = collector.HandleMessage
+
+	state := &stateRecorder{}
+	// the view changer is started so that it can be told about the view it learned
+	viewChanger := &ViewChanger{
+		N:             4,
+		NodesList:     []uint64{1, 2, 3, 4},
+		InMsqQSize:    100,
+		Logger:        log,
+		RequestsTimer: &stubRequestsTimer{},
+	}
+	viewChanger.Start(controllerView)
+	t.Cleanup(viewChanger.Stop)
+
+	c := &Controller{
+		ID:                  1,
+		N:                   4,
+		NodesList:           []uint64{1, 2, 3, 4},
+		Logger:              log,
+		Comm:                comm,
+		Synchronizer:        &stubSynchronizer{response: syncResponse},
+		Checkpoint:          checkpoint,
+		Collector:           collector,
+		InFlight:            &InFlightData{},
+		ViewChanger:         viewChanger,
+		State:               state,
+		currViewNumber:      controllerView,
+		currDecisionsInView: controllerDecisions + 1,
+	}
+	c.syncChan = make(chan struct{}, 1)
+
+	viewNum, _, _ := c.sync()
+
+	assert.Equal(t, clusterView, viewNum,
+		"the node did not learn from its peers that the cluster moved to a higher view")
+
+	// and the node is told about it, so the view changer can move it there
+	assert.NotNil(t, state.saved, "the node did not persist the view it learned from its peers")
+	assert.Equal(t, clusterView, state.saved.GetNewView().ViewId)
+}

internal/bft/util.go

@@ -135,12 +135,16 @@ func (vs *voteSet) registerVote(voter uint64, message *protos.Message) {
 	vs.votes <- &vote{Message: message, sender: voter}
 }
 
+// nextViews tracks the highest view each node announced in a view change message, and the
+// highest view this node already rebroadcast on behalf of each of them.
 type nextViews struct {
-	n map[uint64]uint64
+	n      map[uint64]uint64
+	helped map[uint64]uint64
 }
 
 func (nv *nextViews) clear() {
 	nv.n = make(map[uint64]uint64)
+	nv.helped = make(map[uint64]uint64)
 }
 
 func (nv *nextViews) registerNext(next uint64, sender uint64) {
@@ -155,6 +159,18 @@ func (nv *nextViews) sendRecv(next uint64, sender uint64) bool {
 	return next == nv.n[sender]
 }
 
+// shouldHelp reports whether this node should rebroadcast the view change a lagging node
+// announced for the given view. Repeating the same announcement back only adds noise that the
+// lagging node cannot act on anyway, so each view is helped at most once per node.
+func (nv *nextViews) shouldHelp(next uint64, sender uint64) bool {
+	if next <= nv.helped[sender] {
+		return false
+	}
+
+	nv.helped[sender] = next
+	return true
+}
+
 type incMsg struct {
 	*protos.Message
 	sender uint64

internal/bft/viewchanger.go

@@ -83,15 +83,17 @@ type ViewChanger struct {
 	inFlightView       *View
 	inFlightViewLock   sync.RWMutex
 
-	Ticker              <-chan time.Time
-	lastTick            time.Time
-	ResendTimeout       time.Duration
-	lastResend          time.Time
-	ViewChangeTimeout   time.Duration
-	startViewChangeTime time.Time
-	checkTimeout        bool
-	backOffFactor       uint64
-	syncedForNewView    bool
+	Ticker                    <-chan time.Time
+	lastTick                  time.Time
+	ResendTimeout             time.Duration
+	lastResend                time.Time
+	ViewChangeTimeout         time.Duration
+	startViewChangeTime       time.Time
+	checkTimeout              bool
+	backOffFactor             uint64
+	syncedForNewView          bool
+	syncedForAnnouncedView    uint64
+	hasSyncedForAnnouncedView bool
 
 	// Runtime
 	MetricsViewChange         *api.MetricsViewChange
@@ -240,15 +242,19 @@ func (v *ViewChanger) checkIfResendViewChange(now time.Time) {
 		return
 	}
 	if v.checkTimeout { // during view change process
+		// Announce the view after the one this node is in, since that is the view the other
+		// nodes count. Announcing the view this node is already in would be equal to their
+		// current view, so none of them could count it.
+		nextView := v.currView + 1
 		msg := &protos.Message{
 			Content: &protos.Message_ViewChange{
 				ViewChange: &protos.ViewChange{
-					NextView: v.nextView,
+					NextView: nextView,
 				},
 			},
 		}
 		v.Comm.BroadcastConsensus(msg)
-		v.Logger.Debugf("Node %d resent a view change message with next view %d", v.SelfID, v.nextView)
+		v.Logger.Debugf("Node %d resent a view change message with next view %d", v.SelfID, nextView)
 		v.lastResend = now // update last resend time, or at least last time we checked if we should resend
 	}
 }
@@ -282,10 +288,26 @@ func (v *ViewChanger) processMsg(sender uint64, m *protos.Message) {
 			v.processViewChangeMsg(false)
 			return
 		}
+		if vc.NextView > v.currView+1 {
+			// The sender wants to change into a view that is more than one ahead of this node.
+			// Either this node is behind and has to catch up, or the sender is byzantine.
+			// Syncing resolves both cases: if this node is behind it learns the current view
+			// and moves to it, and if the sender is lying nothing changes.
+			// It syncs at most once per announced view, so that a faulty node cannot make
+			// this node sync over and over again.
+			if !v.hasSyncedForAnnouncedView || v.syncedForAnnouncedView != vc.NextView {
+				v.hasSyncedForAnnouncedView = true
+				v.syncedForAnnouncedView = vc.NextView
+				v.Logger.Debugf("Node %d got a view change message for view %d, while it is in view %d, syncing", v.SelfID, vc.NextView, v.currView)
+				v.Synchronizer.Sync()
+			}
+			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 &&
-			v.nvs.sendRecv(vc.NextView, sender) {
+			v.nvs.sendRecv(vc.NextView, sender) &&
+			v.nvs.shouldHelp(vc.NextView, sender) {
 			// Let's help the lagging nodes.
 			msg := &protos.Message{
 				Content: &protos.Message_ViewChange{
@@ -352,6 +374,8 @@ func (v *ViewChanger) informNewView(view uint64) {
 	v.checkTimeout = false
 	v.backOffFactor = 1 // reset
 	v.syncedForNewView = false
+	v.hasSyncedForAnnouncedView = false // reset
+	v.syncedForAnnouncedView = 0
 	v.RequestsTimer.RestartTimers()
 }
 
@@ -422,8 +446,10 @@ 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.viewDataMsgs.clear(v.N)           // clear because currView changed
+	v.syncedForNewView = false          // clear because currView changed
+	v.hasSyncedForAnnouncedView = false // clear because currView changed
+	v.syncedForAnnouncedView = 0
 	msg := v.prepareViewDataMsg()
 	leader := v.getLeader()
 	if leader == v.SelfID {