diff --git a/internal/bft/controller.go b/internal/bft/controller.go index 4a9eac58..677be103 100644 --- a/internal/bft/controller.go +++ b/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{ diff --git a/internal/bft/controller_sync_test.go b/internal/bft/controller_sync_test.go index b0d0baad..c0861d6c 100644 --- a/internal/bft/controller_sync_test.go +++ b/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) +} diff --git a/internal/bft/util.go b/internal/bft/util.go index 4306b00a..6cfcb8af 100644 --- a/internal/bft/util.go +++ b/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 diff --git a/internal/bft/viewchanger.go b/internal/bft/viewchanger.go index 7e128dad..09cead78 100644 --- a/internal/bft/viewchanger.go +++ b/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 { diff --git a/internal/bft/viewchanger_test.go b/internal/bft/viewchanger_test.go index eff0e8f3..4b1bad51 100644 --- a/internal/bft/viewchanger_test.go +++ b/internal/bft/viewchanger_test.go @@ -163,6 +163,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").Return() vc := &bft.ViewChanger{ SelfID: 0, @@ -178,6 +180,7 @@ func TestViewChangeProcess(t *testing.T) { Controller: controller, InMsqQSize: 100, State: state, + Synchronizer: synchronizer, SpeedUpViewChange: testCase.speedup, } @@ -2686,3 +2689,434 @@ func TestNewViewWithoutLeaderViewData(t *testing.T) { }) } } + +// viewChangeRig is a view changer wired with the mocks it needs, which records the +// consensus messages it sends and broadcasts, the views it completed a view change +// into, and the log entries it wrote, so that a test can inspect all of it. +type viewChangeRig struct { + vc *bft.ViewChanger + controller *mocks.ViewController + ticker chan time.Time + t *testing.T + + mu sync.Mutex + broadcasted []*protos.Message + sent []sentMessage + logged []string + completed []uint64 + syncs int +} + +type sentMessage struct { + to uint64 + msg *protos.Message +} + +func newViewChangeRig(t *testing.T, selfID uint64, nodes []uint64) *viewChangeRig { + t.Helper() + + basicLog, err := zap.NewDevelopment() + assert.NoError(t, err) + + r := &viewChangeRig{t: t} + + logger := basicLog.WithOptions(zap.Hooks(func(entry zapcore.Entry) error { + r.mu.Lock() + defer r.mu.Unlock() + r.logged = append(r.logged, entry.Message) + return nil + })).Sugar() + + comm := &mocks.CommMock{} + comm.On("BroadcastConsensus", mock.Anything).Run(func(args mock.Arguments) { + r.mu.Lock() + defer r.mu.Unlock() + r.broadcasted = append(r.broadcasted, args.Get(0).(*protos.Message)) + }) + comm.On("SendConsensus", mock.Anything, mock.Anything).Run(func(args mock.Arguments) { + r.mu.Lock() + defer r.mu.Unlock() + r.sent = append(r.sent, sentMessage{to: args.Get(0).(uint64), msg: args.Get(1).(*protos.Message)}) + }) + comm.On("Nodes").Return(nodes) + + signer := &mocks.SignerMock{} + signer.On("Sign", mock.Anything).Return([]byte{1, 2, 3}) + + verifier := &mocks.VerifierMock{} + verifier.On("VerifySignature", mock.Anything).Return(nil) + verifier.On("VerifyConsenterSig", mock.Anything, mock.Anything).Return(nil, nil) + verifier.On("RequestsFromProposal", mock.Anything).Return(nil) + verifier.On("VerificationSequence").Return(uint64(0)) + + app := &mocks.ApplicationMock{} + app.On("Deliver", mock.Anything, mock.Anything).Return(types.Reconfig{}) + + synchronizer := &mocks.Synchronizer{} + synchronizer.On("Sync").Run(func(mock.Arguments) { + r.mu.Lock() + defer r.mu.Unlock() + r.syncs++ + }).Return() + + state := &mocks.State{} + state.On("Save", mock.Anything).Return(nil) + + reqTimer := &mocks.RequestsTimer{} + reqTimer.On("StopTimers") + reqTimer.On("RestartTimers") + reqTimer.On("RemoveRequest", mock.Anything).Return(nil) + + r.controller = &mocks.ViewController{} + r.controller.On("AbortView", mock.Anything) + r.controller.On("ViewChanged", mock.Anything, mock.Anything).Run(func(args mock.Arguments) { + r.mu.Lock() + defer r.mu.Unlock() + r.completed = append(r.completed, args.Get(0).(uint64)) + }) + + pruner := &mocks.Pruner{} + pruner.On("MaybePruneRevokedRequests") + + checkpoint := types.Checkpoint{} + checkpoint.Set(lastDecision, lastDecisionSignatures) + + r.ticker = make(chan time.Time) + + r.vc = &bft.ViewChanger{ + SelfID: selfID, + N: uint64(len(nodes)), + NodesList: nodes, + LeaderRotation: false, + DecisionsPerLeader: 0, + Logger: logger, + Comm: comm, + Signer: signer, + Verifier: verifier, + Application: app, + Synchronizer: synchronizer, + State: state, + Checkpoint: &checkpoint, + InFlight: &bft.InFlightData{}, + Controller: r.controller, + RequestsTimer: reqTimer, + Pruner: pruner, + Ticker: r.ticker, + InMsqQSize: 100, + ResendTimeout: time.Millisecond, + ViewChangeTimeout: time.Millisecond, + } + + return r +} + +// forgetRecorded discards everything that was recorded so far, so that a test can +// assert on what happens from this point onwards. +func (r *viewChangeRig) forgetRecorded() { + r.mu.Lock() + defer r.mu.Unlock() + r.broadcasted = nil + r.sent = nil + r.completed = nil +} + +func (r *viewChangeRig) viewChangeBroadcastsTo(view uint64) int { + r.mu.Lock() + defer r.mu.Unlock() + count := 0 + for _, m := range r.broadcasted { + if vc := m.GetViewChange(); vc != nil && vc.NextView == view { + count++ + } + } + return count +} + +func (r *viewChangeRig) newViewBroadcasts() int { + r.mu.Lock() + defer r.mu.Unlock() + count := 0 + for _, m := range r.broadcasted { + if m.GetNewView() != nil { + count++ + } + } + return count +} + +func (r *viewChangeRig) sentMessages() []sentMessage { + r.mu.Lock() + defer r.mu.Unlock() + return append([]sentMessage(nil), r.sent...) +} + +func (r *viewChangeRig) completedViews() []uint64 { + r.mu.Lock() + defer r.mu.Unlock() + return append([]uint64(nil), r.completed...) +} + +func (r *viewChangeRig) syncCount() int { + r.mu.Lock() + defer r.mu.Unlock() + return r.syncs +} + +func (r *viewChangeRig) logCountContaining(sub string) int { + r.mu.Lock() + defer r.mu.Unlock() + count := 0 + for _, l := range r.logged { + if strings.Contains(l, sub) { + count++ + } + } + return count +} + +// waitForLog blocks until the node logged a message containing sub, so that a test does not +// race with the messages the view changer is processing in the background. +func (r *viewChangeRig) waitForLog(sub string) { + r.mu.Lock() + seen := 0 + for _, l := range r.logged { + if strings.Contains(l, sub) { + seen++ + } + } + r.mu.Unlock() + assert.Eventually(r.t, func() bool { + return r.logCountContaining(sub) > seen + }, 3*time.Second, 10*time.Millisecond, "the node never logged a message containing %q", sub) +} + +func viewChangeFor(view uint64) *protos.Message { + return &protos.Message{ + Content: &protos.Message_ViewChange{ + ViewChange: &protos.ViewChange{NextView: view}, + }, + } +} + +// TestViewChangeLaggingNodeCatchesUpAfterViewSkip shows that a node that is behind can join the +// cluster again after the cluster skipped a view. +// +// The node that skipped a view reached its current view through a synchronization, which only +// tells the view changer about a higher view number, it does not run the view change of the +// views in between. The lagging node still has to go through those views one by one, and the +// leader of the view it needs to complete has already moved on, so it rejects the view data and +// no newView message for that view is ever formed. The lagging node then hears the rest of the +// cluster announce a view beyond it, and syncs its way to them instead of getting stuck. +// +// The view numbers below are small, what matters is the gap of two between the lagging node and +// the rest of the cluster, which is what a node is left with when the cluster moved on while it +// was restarting or losing messages. +func TestViewChangeLaggingNodeCatchesUpAfterViewSkip(t *testing.T) { + nodes := []uint64{0, 1, 2, 3} + + // the lagging node, at view 1, while the rest of the cluster is at view 3 + lagging := newViewChangeRig(t, 0, nodes) + lagging.vc.Start(1) + defer lagging.vc.Stop() + + // node 2 is the leader of view 2, the view the lagging node needs, but it already moved on + // to view 3, which is what happens when a node is told about a higher view through a + // synchronization instead of taking part in the view change of view 2 + leader := newViewChangeRig(t, 2, nodes) + leader.vc.Start(3) + defer leader.vc.Stop() + + // the lagging node gathers a quorum of view change messages for view 2, moves to view 2, + // and sends its view data to the leader of view 2, which is node 2 + lagging.vc.HandleMessage(1, viewChangeFor(2)) + lagging.vc.HandleMessage(3, viewChangeFor(2)) + + assert.Eventually(t, func() bool { + return len(lagging.sentMessages()) == 1 + }, 3*time.Second, 10*time.Millisecond, "the lagging node did not send its view data") + + viewData := lagging.sentMessages()[0] + assert.Equal(t, uint64(2), viewData.to, "the view data was not sent to the leader of view 2") + + // the leader of view 2 rejects it, since it is not the leader of view 3 anymore, so the + // lagging node cannot complete view 2 by changing into it + leader.vc.HandleMessage(0, viewData.msg) + assert.Eventually(t, func() bool { + return leader.logCountContaining("is not the next leader of view 3") == 1 + }, 3*time.Second, 10*time.Millisecond, "the leader of the skipped view did not reject the view data") + + assert.Equal(t, 0, leader.newViewBroadcasts(), "a new view message was formed for the skipped view") + assert.Empty(t, lagging.completedViews(), "the lagging node completed the skipped view") + + lagging.forgetRecorded() + + // the rest of the cluster announces its own view change, from view 3 to view 4, which is + // beyond the view the lagging node is in, so the lagging node realizes it is behind + for i := 0; i < 5; i++ { + lagging.vc.HandleMessage(1, viewChangeFor(4)) + lagging.vc.HandleMessage(3, viewChangeFor(4)) + } + + assert.Eventually(t, func() bool { + return lagging.syncCount() > 0 + }, 3*time.Second, 10*time.Millisecond, "the lagging node did not sync when it heard of a view beyond it") + + // the sync learns the current view and the node is told about it, which is how it catches up + lagging.vc.InformNewView(3) + lagging.waitForLog("was informed of a new view 3") + + // once it is in view 3 it takes part in the view change to view 4 like everyone else + lagging.vc.HandleMessage(1, viewChangeFor(4)) + lagging.vc.HandleMessage(3, viewChangeFor(4)) + + // node 0 leads view 4, so it keeps its own view data rather than sending it, but it does + // announce the view change, which is what it could not do while it was stuck behind + assert.Eventually(t, func() bool { + return lagging.viewChangeBroadcastsTo(4) >= 1 + }, 3*time.Second, 10*time.Millisecond, "the lagging node did not join the view change of view 4") + + assert.Eventually(t, func() bool { + return lagging.logCountContaining("sent view data msg, with next view 4") == 1 + }, 3*time.Second, 10*time.Millisecond, "the lagging node did not prepare the view data of view 4") + + // it is no longer announcing the view it is already in, which no node could ever count + assert.Equal(t, 0, lagging.viewChangeBroadcastsTo(3), + "the node announced a view change for the view it is already in, which no node can count") +} + +// TestStaleViewChangeIsHelpedOnlyOnce shows that a node that moved ahead helps a lagging node +// once per view it announces, instead of rebroadcasting the same view change back on every +// message it receives, which is what floods the logs and the network with no new information. +func TestStaleViewChangeIsHelpedOnlyOnce(t *testing.T) { + nodes := []uint64{0, 1, 2, 3} + + rig := newViewChangeRig(t, 2, nodes) + rig.vc.Start(1) + defer rig.vc.Stop() + + // the node goes from view 1 into view 2 without ever completing either of them, so it holds + // a view whose real view is still view 1 + rig.vc.HandleMessage(1, viewChangeFor(2)) + rig.vc.HandleMessage(3, viewChangeFor(2)) + + // wait for the node to actually be in view 2 before asking it to change view again + rig.waitForLog("sent view data msg, with next view 2") + + // and then it starts a view change into view 3 + rig.vc.StartViewChange(2, false) + assert.Eventually(t, func() bool { + return rig.viewChangeBroadcastsTo(3) == 1 + }, 3*time.Second, 10*time.Millisecond, "the node did not start a view change into view 3") + + rig.forgetRecorded() + + // a node that is still in view 1 keeps announcing a view change for view 2 + const laggingMessages = 5 + for i := 0; i < laggingMessages; i++ { + rig.vc.HandleMessage(3, viewChangeFor(2)) + } + + assert.Eventually(t, func() bool { + return rig.logCountContaining("help the lagging nodes") == 1 + }, 3*time.Second, 10*time.Millisecond, "the node did not help the lagging node at all") + + assert.Equal(t, 1, rig.viewChangeBroadcastsTo(2), + "the node rebroadcast the same view change of the lagging node over and over again") + + assert.Empty(t, rig.completedViews(), "the node completed a view change although it never received a new view message") +} + +// TestViewChangeResendAnnouncesNextView shows that a node that changed into a view does not +// keep announcing a view change for the view it is already in. Such an announcement is equal +// to the current view of every other node, so no node can count it and it is pure noise. +func TestViewChangeResendAnnouncesNextView(t *testing.T) { + nodes := []uint64{0, 1, 2, 3} + + rig := newViewChangeRig(t, 0, nodes) + rig.vc.Start(1) + defer rig.vc.Stop() + + // the node changes into view 2 + rig.vc.HandleMessage(1, viewChangeFor(2)) + rig.vc.HandleMessage(3, viewChangeFor(2)) + rig.waitForLog("sent view data msg, with next view 2") + + rig.forgetRecorded() + + // it is still in the middle of the view change, so it resends, and the announcement must + // be for the view after the one it is in + for i := 0; i < 3; i++ { + rig.ticker <- time.Now() + } + + assert.Eventually(t, func() bool { + return rig.viewChangeBroadcastsTo(3) >= 1 + }, 3*time.Second, 10*time.Millisecond, "the node did not announce a view change after the one it is in") + + assert.Equal(t, 0, rig.viewChangeBroadcastsTo(2), + "the node announced a view change for the view it is already in, which no node can count") +} + +// TestRestartedNodeJoinsViaSync covers the case where the cluster did not skip a view on +// purpose, but a node lost the view change it was part of. The node restarts from the view +// it persisted, which is behind the view the cluster moved to, and from there it cannot +// gather a quorum for the view it is in: the nodes ahead only announce the view after the one +// it is in, never the one it needs. It has to notice it is behind and sync its way back in. +func TestRestartedNodeJoinsViaSync(t *testing.T) { + nodes := []uint64{0, 1, 2, 3} + + // node 0 restarts from view 1, the view it persisted before it went down + restarted := newViewChangeRig(t, 0, nodes) + restarted.vc.Start(1) + defer restarted.vc.Stop() + + // the rest of the cluster made it to view 3 without node 0 + restarted.vc.HandleMessage(1, viewChangeFor(4)) + restarted.vc.HandleMessage(3, viewChangeFor(4)) + + // the node cannot count either announcement, they are for a view beyond the one it is in, + // so it has to sync to find out where the cluster actually is + assert.Eventually(t, func() bool { + return restarted.syncCount() > 0 + }, 3*time.Second, 10*time.Millisecond, + "the node did not sync when it heard of a view beyond it, so it cannot rejoin") + + // it does not spin syncing on the same information, one announcement is enough + assert.LessOrEqual(t, restarted.syncCount(), 2, + "the node kept syncing on the same view change announcements") + + // once the sync tells it the cluster is at view 3, it is told about it and is back in + restarted.vc.InformNewView(3) + restarted.waitForLog("was informed of a new view 3") + + restarted.forgetRecorded() + + // and it takes part in the view change to view 4, like any other node + restarted.vc.HandleMessage(1, viewChangeFor(4)) + restarted.vc.HandleMessage(3, viewChangeFor(4)) + + assert.Eventually(t, func() bool { + return restarted.viewChangeBroadcastsTo(4) >= 1 + }, 3*time.Second, 10*time.Millisecond, "the restarted node did not join the view change of view 4") +} + +// TestRepeatedDistantViewChangeSyncsOnce makes sure a faulty node cannot make this node sync +// over and over again by repeatedly announcing a view that is far ahead of it. +func TestRepeatedDistantViewChangeSyncsOnce(t *testing.T) { + nodes := []uint64{0, 1, 2, 3} + + rig := newViewChangeRig(t, 0, nodes) + rig.vc.Start(1) + defer rig.vc.Stop() + + for i := 0; i < 10; i++ { + rig.vc.HandleMessage(1, viewChangeFor(50)) + } + + assert.Eventually(t, func() bool { + return rig.syncCount() > 0 + }, 3*time.Second, 10*time.Millisecond, "the node did not sync when it heard of a view far beyond it") + + assert.LessOrEqual(t, rig.syncCount(), 2, + "the node kept syncing on every message, a faulty node can make it sync indefinitely this way") +}