ibft: fix edge case where validators could get stuck on round-change …

…requests on high txs

This fixes a specific edge case where validators could get stuck
producing blocks and unable to recover through round-change when on high
transactions throughput and one or more validators fail to send
pre-prepare in time.

consensus/istanbul/core/core.go

@@ -53,6 +53,7 @@ func New(backend istanbul.Backend, config *istanbul.Config) istanbul.Core {
 		pendingRequests:    prque.New(),
 		pendingRequestsMu:  new(sync.Mutex),
 		consensusTimestamp: time.Time{},
+		currentMutex:       new(sync.Mutex),
 	}
 
 	c.validateFn = c.checkValidatorSignature
@@ -80,8 +81,9 @@ type core struct {
 	backlogs   map[common.Address]*prque.Prque
 	backlogsMu *sync.Mutex
 
-	current   *roundState
-	handlerWg *sync.WaitGroup
+	current      *roundState
+	currentMutex *sync.Mutex
+	handlerWg    *sync.WaitGroup
 
 	roundChangeSet   *roundChangeSet
 	roundChangeTimer *time.Timer
@@ -115,6 +117,10 @@ func (c *core) IsCurrentProposal(blockHash common.Hash) bool {
 
 // startNewRound starts a new round. if round equals to 0, it means to starts a new sequence
 func (c *core) startNewRound(round *big.Int) {
+	c.currentMutex.Lock()
+	defer c.currentMutex.Unlock()
+	defer c.newRoundChangeTimer()
+
 	var logger log.Logger
 	if c.current == nil {
 		logger = c.logger.New("old.round", -1, "old.seq", 0)
@@ -206,10 +212,6 @@ func (c *core) startNewRound(round *big.Int) {
 	}
 	c.roundChangeSet.NewRound(round)
 
-	if round.Uint64() > 0 {
-		c.newRoundChangeTimer()
-	}
-
 	oldLogger.Trace("IBFT: start new round", "next.round", newView.Round, "next.seq", newView.Sequence, "next.proposer", c.valSet.GetProposer(), "next.valSet", c.valSet.List(), "next.size", c.valSet.Size(), "next.IsProposer", c.IsProposer())
 }
 

consensus/istanbul/core/preprepare.go

@@ -34,6 +34,10 @@ import (
 // - extends PRE-PREPARE message with ROUND-CHANGE and PREPARE justification
 // - broadcast PRE-PREPARE message to other validators
 func (c *core) sendPreprepareMsg(request *Request) {
+	// c.current and c.valSet (checked in IsProposer()) is updated asynchronously in startNewRound(),
+	// need to prevent race condition with mutex
+	c.currentMutex.Lock()
+	defer c.currentMutex.Unlock()
 	logger := c.currentLogger(true, nil)
 
 	// If I'm the proposer and I have the same sequence with the proposal
@@ -95,10 +99,6 @@ func (c *core) sendPreprepareMsg(request *Request) {
 		// Set the preprepareSent to the current round
 		c.current.preprepareSent = curView.Round
 	}
-
-	if !c.IsProposer() {
-		c.cleanLogger.Info("Consensus: Handling incoming block proposal", "sequence", c.current.Sequence().Uint64(), "author", c.valSet.GetProposer())
-	}
 }
 
 // handlePreprepareMsg is called when receiving a PRE-PREPARE message from the proposer
@@ -154,8 +154,7 @@ func (c *core) handlePreprepareMsg(preprepare *qbfttypes.Preprepare) error {
 	if c.state == StateAcceptRequest {
 		c.logger.Trace("IBFT: accepted PRE-PREPARE message")
 
-		// Re-initialize ROUND-CHANGE timer
-		c.newRoundChangeTimer()
+		// Update consensus timestamp
 		c.consensusTimestamp = time.Now()
 
 		// Update current state

consensus/istanbul/core/request.go

@@ -40,9 +40,6 @@ func (c *core) handleRequest(request *Request) error {
 
 	c.current.pendingRequest = request
 	if c.state == StateAcceptRequest {
-		// Start ROUND-CHANGE timer
-		c.newRoundChangeTimer()
-
 		// Send PRE-PREPARE message to other validators
 		c.sendPreprepareMsg(request)
 	}

consensus/istanbul/core/roundchange.go

@@ -163,7 +163,7 @@ func (c *core) handleRoundChange(roundChange *qbfttypes.RoundChange) error {
 		// If we have received a quorum of PREPARE messages with hadBlockProposal=false,
 		// propose the same block again. If hadBlockProposal=true, propose the block that we generated
 		_, proposal := c.highestPrepared(currentRound)
-		if proposal == nil {
+		if proposal == nil || c.backend.HasBadProposal(proposal.Hash()) {
 			if c.current != nil && c.current.pendingRequest != nil {
 				proposal = c.current.pendingRequest.Proposal
 			} else {