Properly handle ticks when waiting for In-Flight view to finish

Due to a bug, When the view changer initializes an in-flight view it only waits for a single tick
before resuming view change standard operation.

Another part of the bug is that it anyway returns success, which in turn makes the sequence of the controller
be initialized to an old sequence.

Signed-off-by: Yacov Manevich <yacovm@il.ibm.com>

Author: yacovm

internal/bft/viewchanger.go

@@ -233,13 +233,13 @@ func (v *ViewChanger) checkIfResendViewChange(now time.Time) {
 	}
 }
 
-func (v *ViewChanger) checkIfTimeout(now time.Time) {
+func (v *ViewChanger) checkIfTimeout(now time.Time) bool {
 	if !v.checkTimeout {
-		return
+		return false
 	}
 	nextTimeout := v.startViewChangeTime.Add(v.ViewChangeTimeout * time.Duration(v.backOffFactor))
 	if nextTimeout.After(now) { // check if timeout has passed
-		return
+		return false
 	}
 	v.Logger.Debugf("Node %d got a view change timeout, the current view is %d", v.SelfID, v.currView)
 	v.checkTimeout = false // stop timeout for now, a new one will start when a new view change begins
@@ -248,6 +248,7 @@ func (v *ViewChanger) checkIfTimeout(now time.Time) {
 	v.Logger.Debugf("Node %d is calling sync because it got a view change timeout", v.SelfID)
 	v.Synchronizer.Sync()
 	v.StartViewChange(v.currView, false) // don't stop the view, the sync maybe created a good view
+	return true
 }
 
 func (v *ViewChanger) processMsg(sender uint64, m *protos.Message) {
@@ -1175,13 +1176,16 @@ func (v *ViewChanger) commitInFlightProposal(proposal *protos.Proposal) (success
 
 	v.Logger.Debugf("Node %d is creating a view for the in flight proposal", v.SelfID)
 
+	inFlightViewNum := proposalMD.ViewId
+	inFlightViewLatestSeq := proposalMD.LatestSequence
+
 	v.inFlightViewLock.Lock()
-	v.inFlightView = &View{
+	inFlightView := &View{
 		RetrieveCheckpoint: v.Checkpoint.Get,
 		DecisionsPerLeader: v.DecisionsPerLeader,
 		SelfID:             v.SelfID,
 		N:                  v.N,
-		Number:             proposalMD.ViewId,
+		Number:             inFlightViewNum,
 		LeaderID:           v.SelfID, // so that no byzantine leader will cause a complain
 		Quorum:             v.quorum,
 		Decider:            v,
@@ -1191,13 +1195,13 @@ func (v *ViewChanger) commitInFlightProposal(proposal *protos.Proposal) (success
 		Comm:               v.Comm,
 		Verifier:           v.Verifier,
 		Signer:             v.Signer,
-		ProposalSequence:   proposalMD.LatestSequence,
+		ProposalSequence:   inFlightViewLatestSeq,
 		State:              v.State,
 		InMsgQSize:         v.InMsqQSize,
 		ViewSequences:      v.ViewSequences,
 		Phase:              PREPARED,
 	}
-
+	v.inFlightView = inFlightView
 	v.inFlightView.inFlightProposal = &types.Proposal{
 		VerificationSequence: int64(proposal.VerificationSequence),
 		Metadata:             proposal.Metadata,
@@ -1220,21 +1224,34 @@ func (v *ViewChanger) commitInFlightProposal(proposal *protos.Proposal) (success
 		},
 	}
 
-	v.inFlightView.Start()
+	inFlightView.Start()
+	defer inFlightView.Abort()
+
 	v.inFlightViewLock.Unlock()
 
 	v.Logger.Debugf("Node %d started a view for the in flight proposal", v.SelfID)
 
-	select { // wait for view to finish
-	case <-v.inFlightDecideChan:
-	case <-v.inFlightSyncChan:
-	case <-v.stopChan:
-	case now := <-v.Ticker:
-		v.lastTick = now
-		v.checkIfTimeout(now)
+	// wait for view to finish or time out
+	for {
+		select {
+		case <-v.inFlightDecideChan:
+			v.Logger.Infof("In-flight view %d with latest sequence %d has committed a decision", inFlightViewNum, inFlightViewLatestSeq)
+			return true
+		case <-v.inFlightSyncChan:
+			v.Logger.Infof("In-flight view %d with latest sequence %d has asked to sync", inFlightViewNum, inFlightViewLatestSeq)
+			return false
+		case now := <-v.Ticker:
+			v.lastTick = now
+			if v.checkIfTimeout(now) {
+				v.Logger.Infof("Timeout expired waiting on In-flight %d with latest sequence view to commit %d", inFlightViewNum, inFlightViewLatestSeq)
+				return false
+			}
+		case <-v.stopChan:
+			v.Logger.Infof("View changer was instructed to stop")
+			return false
+		}
 	}
 
-	v.inFlightView.Abort()
 	return true
 }
 

test/basic_test.go

@@ -1657,6 +1657,150 @@ func TestMigrateToBlacklistAndBackAgain(t *testing.T) {
 	})
 }
 
+func TestNodeInFlightFails(t *testing.T) {
+	t.Parallel()
+	network := make(Network)
+	defer network.Shutdown()
+
+	testDir, err := ioutil.TempDir("", t.Name())
+	assert.NoErrorf(t, err, "generate temporary test dir")
+	defer os.RemoveAll(testDir)
+
+	numberOfNodes := 4
+	nodes := make([]*App, 0)
+
+	var failedViewChange sync.WaitGroup
+	failedViewChange.Add(1)
+
+	var timeoutExpiredWG sync.WaitGroup
+	timeoutExpiredWG.Add(1)
+
+	var collectPreparesWG sync.WaitGroup
+	collectPreparesWG.Add(4)
+
+	var inFlightCommit sync.WaitGroup
+	inFlightCommit.Add(2)
+
+	var blockCommits uint32
+	var blockCommitsForLastNode uint32
+
+	for i := 1; i <= numberOfNodes; i++ {
+		n := newNode(uint64(i), network, t.Name(), testDir, false, 0)
+		n.Consensus.Config.SyncOnStart = false                       // Prevent last node from syncing without a reason
+		n.Consensus.Config.ViewChangeTimeout = time.Second * 10      // Make view change faster to speedup the test
+		n.Consensus.Config.LeaderHeartbeatTimeout = time.Second * 10 // Force view change sooner to speedup test
+		n.Consensus.Config.RequestForwardTimeout = time.Second * 25  // Prevent first node to forward after view change
+		n.Consensus.Config.RequestComplainTimeout = time.Second * 30 // Must be bigger than the forward timeout
+		n.Consensus.Config.SpeedUpViewChange = false
+		nodes = append(nodes, n)
+
+		baseLogger := n.logger.Desugar()
+
+		id := n.ID
+		n.Consensus.Logger = baseLogger.WithOptions(zap.Hooks(func(entry zapcore.Entry) error {
+			if strings.Contains(entry.Message, "collected 2 prepares") && atomic.LoadUint32(&blockCommits) == 1 {
+				collectPreparesWG.Done()
+			}
+
+			if strings.Contains(entry.Message, "In-flight view 0 with latest sequence 1 has committed a decision") && (id == 2 || id == 3) {
+				inFlightCommit.Done()
+			}
+
+			if strings.Contains(entry.Message, "Timeout expired waiting on In-flight 0 with latest sequence view to commit 1") && id == 4 {
+				timeoutExpiredWG.Done()
+			}
+
+			if strings.Contains(entry.Message, "Node 4 was unable to commit the in flight proposal, not changing the view") && id == 4 {
+				failedViewChange.Done()
+			}
+
+			return nil
+		})).Sugar()
+
+		// Make all nodes but the last node lose commits per the 'blockCommits' flag
+		if i == 4 {
+			continue
+		}
+		n.LoseMessages(func(msg *smartbftprotos.Message) bool {
+			if msg.GetCommit() == nil {
+				return false
+			}
+			return atomic.LoadUint32(&blockCommits) == 1
+		})
+	}
+
+	// However the last node never receives any commits.
+	nodes[len(nodes)-1].LoseMessages(func(msg *smartbftprotos.Message) bool {
+		return msg.GetCommit() != nil && atomic.LoadUint32(&blockCommitsForLastNode) == 1
+	})
+
+	startNodes(nodes, &network)
+
+	atomic.StoreUint32(&blockCommits, 1)
+	atomic.StoreUint32(&blockCommitsForLastNode, 1)
+
+	nodes[0].Submit(Request{ID: "first request", ClientID: "alice"})
+
+	// Wait for prepares to be collected
+	collectPreparesWG.Wait()
+
+	// Ensure no one commits a block because of not enough commits gathered
+	var doNotCommitWG sync.WaitGroup
+	doNotCommitWG.Add(numberOfNodes)
+
+	for i := 0; i < numberOfNodes; i++ {
+		go func(deliverChannel chan *AppRecord) {
+			defer doNotCommitWG.Done()
+			select {
+			case <-deliverChannel:
+				assert.Fail(t, "should not have committed because first commit message should have been lost")
+			case <-time.After(time.Second):
+			}
+		}(nodes[i].Delivered)
+	}
+	doNotCommitWG.Wait()
+
+	// Re-enable commit messages to flow to all nodes but the last node
+	atomic.StoreUint32(&blockCommits, 0)
+
+	// Disconnect leader to force view change
+	nodes[0].Disconnect()
+
+	// Wait for view change and ensure in-flight view has committed the previous decision
+	inFlightCommit.Wait()
+
+	// Re-connect the leader
+	nodes[0].Connect()
+
+	// Ensure that only nodes 1 to 3 have committed in-flight
+	d1 := <-nodes[0].Delivered
+	d2 := <-nodes[1].Delivered
+	d3 := <-nodes[2].Delivered
+	select {
+	case <-nodes[3].Delivered:
+		t.Fatalf("Node 4 has committed but shouldn't have")
+	case <-time.After(time.Second):
+	}
+
+	// Ensure all nodes that committed, committed the same decision
+	assert.Equal(t, d1, d2)
+	assert.Equal(t, d2, d3)
+
+	// Wait for in-flight view of view changer to time out
+	timeoutExpiredWG.Wait()
+	// Re-enable commits to flow to the last node
+	atomic.StoreUint32(&blockCommitsForLastNode, 0)
+	// Ensure the last node failed changing the view
+	failedViewChange.Wait()
+	// However, it eventually syncs successfully and delivers the proposal
+	select {
+	case d4 := <-nodes[3].Delivered:
+		assert.Equal(t, d3, d4)
+	case <-time.After(time.Second * 30):
+		t.Fatalf("Didn't commit decision in a timely manner")
+	}
+}
+
 func TestBlacklistAndRedemption(t *testing.T) {
 	t.Parallel()
 	network := make(Network)

test/network.go

@@ -181,6 +181,9 @@ func (node *Node) serve() {
 			node.RUnlock()
 			switch msg := m.message.(type) {
 			case *smartbftprotos.Message:
+				if node.app != nil && node.app.messageLost != nil && node.app.messageLost(msg) {
+					continue
+				}
 				handler.HandleMessage(uint64(m.from), msg)
 			default:
 				handler.HandleRequest(uint64(m.from), msg.(*FwdMessage).Payload)

test/test_app.go

@@ -59,6 +59,7 @@ type App struct {
 	logger          *zap.SugaredLogger
 	lastRecord      lastRecord
 	verificationSeq uint64
+	messageLost     func(*smartbftprotos.Message) bool
 }
 
 type lastRecord struct {
@@ -180,6 +181,10 @@ func (a *App) ClearMutateSend(target uint64) {
 	delete(a.Node.peerMutatingFunc, target)
 }
 
+func (a *App) LoseMessages(filter func(*smartbftprotos.Message) bool) {
+	a.messageLost = filter
+}
+
 // RequestID returns info about the given request
 func (a *App) RequestID(req []byte) types.RequestInfo {
 	txn := requestFromBytes(req)
@@ -245,7 +250,7 @@ func (a *App) Sign([]byte) []byte {
 
 // SignProposal signs on the given proposal
 func (a *App) SignProposal(_ types.Proposal, aux []byte) *types.Signature {
-	if len(aux) == 0 && len(a.Node.n) > 1 {
+	if len(aux) == 0 && len(a.Node.n) > 1 && a.messageLost == nil {
 		panic(fmt.Sprintf("didn't receive prepares from anyone, n=%d", len(a.Node.n)))
 	}
 	return &types.Signature{ID: a.ID, Msg: aux}