Syncer stuck fix (#255)

* ping best peer after some time

* small fix

* small reorg

* fetch peer statuses if timeout occurrs

* scale blockTimeout in regards to round timeout

consensus/polybft/mocks_test.go

@@ -408,6 +408,10 @@ func (tp *syncerMock) Sync(func(*types.FullBlock) bool) error {
 	return args.Error(0)
 }
 
+func (tp *syncerMock) UpdateBlockTimeout(time.Duration) {
+	tp.Called()
+}
+
 func init() {
 	// setup custom hash header func
 	setupHeaderHashFunc()

consensus/polybft/polybft.go

@@ -101,9 +101,6 @@ type Polybft struct {
 	// runtime handles consensus runtime features like epoch, state and event management
 	runtime *consensusRuntime
 
-	// block time duration
-	blockTime time.Duration
-
 	// dataDir is the data directory to store the info
 	dataDir string
 
@@ -489,9 +486,6 @@ func (p *Polybft) Initialize() error {
 		return fmt.Errorf("cannot create topics: %w", err)
 	}
 
-	// set block time
-	p.blockTime = time.Duration(p.config.BlockTime)
-
 	// initialize polybft consensus data directory
 	p.dataDir = filepath.Join(p.config.Config.Path, "polybft")
 	// create the data dir if not exists
@@ -776,9 +770,13 @@ func (p *Polybft) GetValidatorsWithTx(blockNumber uint64, parents []*types.Heade
 func (p *Polybft) SetBlockTime(blockTime time.Duration) {
 	// if block time is greater than default base round timeout,
 	// set base round timeout as twice the block time
+	syncerBlockTimeout := blockTime * 3
 	if blockTime >= core.DefaultBaseRoundTimeout {
 		p.ibft.SetBaseRoundTimeout(blockTime * baseRoundTimeoutScaleFactor)
+		syncerBlockTimeout *= baseRoundTimeoutScaleFactor
 	}
+
+	p.syncer.UpdateBlockTimeout(syncerBlockTimeout)
 }
 
 // ProcessHeaders updates the snapshot based on the verified headers

syncer/syncer.go

@@ -3,6 +3,7 @@ package syncer
 import (
 	"errors"
 	"fmt"
+	"sync"
 	"time"
 
 	"github.com/0xPolygon/polygon-edge/helper/progress"
@@ -38,6 +39,8 @@ type syncer struct {
 
 	// Channel to notify Sync that a new status arrived
 	newStatusCh chan struct{}
+
+	lock sync.RWMutex
 }
 
 func NewSyncer(
@@ -58,6 +61,14 @@ func NewSyncer(
 	}
 }
 
+// UpdateBlockTimeout updates block timeout in syncer
+func (s *syncer) UpdateBlockTimeout(timeout time.Duration) {
+	s.lock.Lock()
+	defer s.lock.Unlock()
+
+	s.blockTimeout = timeout
+}
+
 // Start starts goroutine processes
 func (s *syncer) Start() error {
 	if err := s.syncPeerClient.Start(); err != nil {
@@ -165,8 +176,18 @@ func (s *syncer) Sync(callback func(*types.FullBlock) bool) error {
 	skipList := make(map[peer.ID]bool)
 
 	for {
+		s.lock.RLock()
+		blockTimeout := s.blockTimeout
+		s.lock.RUnlock()
+
 		// Wait for a new event to arrive
-		<-s.newStatusCh
+		select {
+		case <-s.newStatusCh:
+			s.logger.Debug("new peer status arrived, start syncing")
+		case <-time.After(blockTimeout):
+			s.logger.Debug("timeout while waiting for new peer status, start manual syncing")
+			s.initializePeerMap() // fetch peer statuses just in case
+		}
 
 		// fetch local latest block
 		if header := s.blockchain.Header(); header != nil {
@@ -214,7 +235,11 @@ func (s *syncer) bulkSyncWithPeer(peerID peer.ID, peerLatestBlock uint64,
 	localLatest := s.blockchain.Header().Number
 	shouldTerminate := false
 
-	blockCh, err := s.syncPeerClient.GetBlocks(peerID, localLatest+1, s.blockTimeout)
+	s.lock.RLock()
+	blockTimeout := s.blockTimeout
+	s.lock.RUnlock()
+
+	blockCh, err := s.syncPeerClient.GetBlocks(peerID, localLatest+1, blockTimeout)
 	if err != nil {
 		return 0, false, err
 	}
@@ -266,7 +291,7 @@ func (s *syncer) bulkSyncWithPeer(peerID peer.ID, peerLatestBlock uint64,
 			shouldTerminate = newBlockCallback(fullBlock)
 
 			lastReceivedNumber = block.Number()
-		case <-time.After(s.blockTimeout):
+		case <-time.After(blockTimeout):
 			return lastReceivedNumber, shouldTerminate, errTimeout
 		}
 	}

syncer/types.go

@@ -70,6 +70,8 @@ type Syncer interface {
 	HasSyncPeer() bool
 	// Sync starts routine to sync blocks
 	Sync(func(*types.FullBlock) bool) error
+	// UpdateBlockTimeout updates block timeout in syncer
+	UpdateBlockTimeout(time.Duration)
 }
 
 type Progression interface {