stale sequence check (temp fix)

consensus/polybft/fsm.go

@@ -211,6 +211,8 @@ func (f *fsm) BuildProposal(currentRound uint64) ([]byte, error) {
 		}
 
 		f.logger.Debug("[FSM Build Proposal]",
+			"block", stateBlock.Block.Number(),
+			"round", currentRound,
 			"txs", len(stateBlock.Block.Transactions),
 			"proposal hash", checkpointHash.String())
 	}
@@ -392,7 +394,8 @@ func (f *fsm) Validate(proposal []byte) error {
 			return fmt.Errorf("failed to calculate proposal hash: %w", err)
 		}
 
-		f.logger.Debug("[FSM Validate]", "txs", len(block.Transactions), "proposal hash", checkpointHash)
+		f.logger.Debug("[FSM Validate] Finish", "Block", block.Number(), "Proposer", types.BytesToHash(block.Header.Miner),
+			"txs", len(block.Transactions), "proposal hash", checkpointHash)
 	}
 
 	f.target = stateBlock

consensus/polybft/polybft.go

@@ -629,6 +629,14 @@ func (p *Polybft) startConsensusProtocol() {
 		stopSequence func()
 	)
 
+	// check every block time * 1.5, so we don't artificially close a sequence before the actual sequence has ended properly
+	checkOffset := p.config.BlockTime
+	if checkOffset > 1 {
+		checkOffset = checkOffset / 2
+	}
+	checkFrequency := time.Duration(p.config.BlockTime + checkOffset)
+	staleChecker := newStaleSequenceCheck(p.logger, p.blockchain.CurrentHeader, checkFrequency*time.Second)
+
 	for {
 		latestHeader := p.blockchain.CurrentHeader()
 
@@ -652,6 +660,8 @@ func (p *Polybft) startConsensusProtocol() {
 			}
 
 			sequenceCh, stopSequence = p.ibft.runSequence(latestHeader.Number + 1)
+			staleChecker.setSequence(latestHeader.Number + 1)
+			staleChecker.startChecking()
 		}
 
 		now := time.Now().UTC()
@@ -662,15 +672,26 @@ func (p *Polybft) startConsensusProtocol() {
 				stopSequence()
 				p.logger.Info("canceled sequence", "sequence", latestHeader.Number+1)
 			}
+		case <-staleChecker.sequenceShouldStop:
+			if isValidator {
+				stopSequence()
+				p.logger.Info("canceled sequence via stale checker", "sequence", latestHeader.Number+1)
+			}
 		case <-sequenceCh:
 		case <-p.closeCh:
+			p.logger.Debug("[BRE] stoping sequence", "block number", latestHeader.Number+1)
 			if isValidator {
 				stopSequence()
+				staleChecker.stopChecking()
 			}
 
 			return
 		}
 
+		if isValidator {
+			staleChecker.stopChecking()
+		}
+
 		p.logger.Debug("time to run the sequence", "seconds", time.Since(now))
 	}
 }

consensus/polybft/stale_sequence_check.go

@@ -0,0 +1,79 @@
+package polybft
+
+import (
+	"sync"
+	"time"
+
+	"github.com/0xPolygon/polygon-edge/types"
+	"github.com/hashicorp/go-hclog"
+)
+
+type staleSequenceCheck struct {
+	logger             hclog.Logger
+	currentSequence    uint64
+	mtx                *sync.Mutex
+	checkFrequency     time.Duration
+	sequenceShouldStop chan struct{}
+	stop               chan struct{}
+	stopped            chan struct{}
+	getHeader          func() *types.Header
+}
+
+func newStaleSequenceCheck(logger hclog.Logger, getHeader func() *types.Header, checkDuration time.Duration) *staleSequenceCheck {
+	return &staleSequenceCheck{
+		logger:          logger,
+		currentSequence: 0,
+		mtx:             &sync.Mutex{},
+		checkFrequency:  checkDuration,
+		getHeader:       getHeader,
+	}
+}
+
+func (s *staleSequenceCheck) startChecking() {
+	s.sequenceShouldStop = make(chan struct{}, 1)
+	s.stop = make(chan struct{})
+	s.stopped = make(chan struct{})
+	ticker := time.NewTicker(s.checkFrequency)
+	go func() {
+		defer close(s.stopped)
+		for {
+			select {
+			case <-s.stop:
+				close(s.sequenceShouldStop)
+				ticker.Stop()
+				return
+			case <-ticker.C:
+				s.checkForStaleness()
+			}
+		}
+	}()
+}
+
+func (s *staleSequenceCheck) stopChecking() {
+	close(s.stop)
+	<-s.stopped
+}
+
+func (s *staleSequenceCheck) setSequence(sequence uint64) {
+	s.mtx.Lock()
+	defer s.mtx.Unlock()
+	s.currentSequence = sequence
+}
+
+func (s *staleSequenceCheck) checkForStaleness() {
+	s.logger.Info("[staleSequenceCheck] checking for stale sequence")
+	header := s.getHeader()
+	s.chainHeightUpdated(header.Number)
+}
+
+func (s *staleSequenceCheck) chainHeightUpdated(height uint64) {
+	s.mtx.Lock()
+	defer s.mtx.Unlock()
+	if s.currentSequence == 0 {
+		return
+	}
+	if height >= s.currentSequence {
+		s.logger.Info("[staleSequenceCheck] stale sequence detected", "height", height, "currentSequence", s.currentSequence)
+		s.sequenceShouldStop <- struct{}{}
+	}
+}