From c690e2244ce40d4becec7399d730ff41c3291355 Mon Sep 17 00:00:00 2001 From: Fedor Partanskiy Date: Sat, 10 Oct 2026 16:15:28 +0300 Subject: [PATCH] fix: let a node move on when the cluster is more than one sequence ahead A node that lost a view change, for example by restarting, stays in that view while the rest of the cluster moved on. It can never rejoin: the view change only accepts messages for the view right after the one the node is in, and no node announces that view anymore. It is told to sync, but sync only accepted the state the peers reported when they were exactly one sequence ahead, so as soon as the cluster decided more than one transaction the node discarded the view it was given and kept where it was, forever. Accept any sequence the cluster is ahead by, the synchronizer catches up the decisions anyway. A node that hears about a view beyond the one it is in now syncs to find out where the cluster is, at most once per announced view so a faulty node cannot make it sync indefinitely. Its view change resend announces the view after the one it is in, since announcing the view it is already in cannot be counted by any node. Lagging nodes are helped once per view instead of on every message, which is what flooded the logs. Signed-off-by: Fedor Partanskiy --- internal/bft/controller.go | 7 +- internal/bft/controller_sync_test.go | 157 ++++++++++ internal/bft/util.go | 18 +- internal/bft/viewchanger.go | 54 +++- internal/bft/viewchanger_test.go | 434 +++++++++++++++++++++++++++ 5 files changed, 654 insertions(+), 16 deletions(-) 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") +}