Xin 231 sync issues - too far messages should omit and skip to process (#177)

* omit too far v2 messages

* update error msg

* improve log

* fix test

* remove useless file

consensus/XDPoS/engines/engine_v2/engine.go

@@ -27,6 +27,7 @@ type XDPoS_v2 struct {
 	config       *params.XDPoSConfig // Consensus engine configuration parameters
 	db           ethdb.Database      // Database to store and retrieve snapshot checkpoints
 	isInitilised bool                // status of v2 variables
+	whosTurn     common.Address      // Record waiting for who to mine
 
 	snapshots       *lru.ARCCache // Snapshots for gap block
 	signatures      *lru.ARCCache // Signatures of recent blocks to speed up mining
@@ -525,12 +526,12 @@ func (x *XDPoS_v2) VerifySyncInfoMessage(chain consensus.ChainReader, syncInfo *
 
 	err := x.verifyQC(chain, syncInfo.HighestQuorumCert, nil)
 	if err != nil {
-		log.Warn("[VerifySyncInfoMessage] SyncInfo message verification failed due to QC", "error", err)
+		log.Warn("[VerifySyncInfoMessage] SyncInfo message verification failed due to QC", "blockNum", syncInfo.HighestQuorumCert.ProposedBlockInfo.Number, "round", syncInfo.HighestQuorumCert.ProposedBlockInfo.Round, "error", err)
 		return false, err
 	}
 	err = x.verifyTC(chain, syncInfo.HighestTimeoutCert)
 	if err != nil {
-		log.Warn("[VerifySyncInfoMessage] SyncInfo message verification failed due to TC", "error", err)
+		log.Warn("[VerifySyncInfoMessage] SyncInfo message verification failed due to TC", "gapNum", syncInfo.HighestTimeoutCert.GapNumber, "round", syncInfo.HighestTimeoutCert.Round, "error", err)
 		return false, err
 	}
 	return true, nil
@@ -564,13 +565,13 @@ func (x *XDPoS_v2) VerifyVoteMessage(chain consensus.ChainReader, vote *types.Vo
 			4. Broadcast(Not part of consensus)
 	*/
 	if vote.ProposedBlockInfo.Round < x.currentRound {
-		log.Debug("[VerifyVoteMessage] Disqualified vote message as the proposed round does not match currentRound", "vote.ProposedBlockInfo.Round", vote.ProposedBlockInfo.Round, "currentRound", x.currentRound)
+		log.Debug("[VerifyVoteMessage] Disqualified vote message as the proposed round does not match currentRound", "voteHash", vote.Hash(), "voteProposedBlockInfoRound", vote.ProposedBlockInfo.Round, "currentRound", x.currentRound)
 		return false, nil
 	}
 
 	snapshot, err := x.getSnapshot(chain, vote.GapNumber, true)
 	if err != nil {
-		log.Error("[VerifyVoteMessage] fail to get snapshot for a vote message", "BlockNum", vote.ProposedBlockInfo.Number, "Hash", vote.ProposedBlockInfo.Hash, "Error", err.Error())
+		log.Error("[VerifyVoteMessage] fail to get snapshot for a vote message", "blockNum", vote.ProposedBlockInfo.Number, "blockHash", vote.ProposedBlockInfo.Hash, "voteHash", vote.Hash(), "error", err.Error())
 		return false, err
 	}
 	verified, _, err := x.verifyMsgSignature(types.VoteSigHash(&types.VoteForSign{
@@ -581,7 +582,7 @@ func (x *XDPoS_v2) VerifyVoteMessage(chain consensus.ChainReader, vote *types.Vo
 		for i, mn := range snapshot.NextEpochMasterNodes {
 			log.Warn("[VerifyVoteMessage] Master node list item", "index", i, "Master node", mn.Hex())
 		}
-		log.Warn("[VerifyVoteMessage] Error while verifying vote message", "votedBlockNum", vote.ProposedBlockInfo.Number.Uint64(), "votedBlockHash", vote.ProposedBlockInfo.Hash.Hex(), "Error", err.Error())
+		log.Warn("[VerifyVoteMessage] Error while verifying vote message", "votedBlockNum", vote.ProposedBlockInfo.Number.Uint64(), "votedBlockHash", vote.ProposedBlockInfo.Hash.Hex(), "voteHash", vote.Hash(), "error", err.Error())
 	}
 	return verified, err
 }
@@ -875,7 +876,7 @@ func (x *XDPoS_v2) commitBlocks(blockChainReader consensus.ChainReader, proposed
 		return false, err
 	}
 	if *proposedBlockRound-1 != round {
-		log.Info("[commitBlocks] Rounds not continuous(parent) found when committing block", "proposedBlockRound", proposedBlockRound, "decodedExtraField.Round", round, "proposedBlockHeaderHash", proposedBlockHeader.Hash())
+		log.Info("[commitBlocks] Rounds not continuous(parent) found when committing block", "proposedBlockRound", *proposedBlockRound, "decodedExtraField.Round", round, "proposedBlockHeaderHash", proposedBlockHeader.Hash())
 		return false, nil
 	}
 
@@ -887,7 +888,7 @@ func (x *XDPoS_v2) commitBlocks(blockChainReader consensus.ChainReader, proposed
 		return false, err
 	}
 	if *proposedBlockRound-2 != round {
-		log.Info("[commitBlocks] Rounds not continuous(grand parent) found when committing block", "proposedBlockRound", proposedBlockRound, "decodedExtraField.Round", round, "proposedBlockHeaderHash", proposedBlockHeader.Hash())
+		log.Info("[commitBlocks] Rounds not continuous(grand parent) found when committing block", "proposedBlockRound", *proposedBlockRound, "decodedExtraField.Round", round, "proposedBlockHeaderHash", proposedBlockHeader.Hash())
 		return false, nil
 	}
 

consensus/XDPoS/engines/engine_v2/mining.go

@@ -51,8 +51,9 @@ func (x *XDPoS_v2) yourturn(chain consensus.ChainReader, round types.Round, pare
 	}
 
 	leaderIndex := uint64(round) % x.config.Epoch % uint64(len(masterNodes))
-	if masterNodes[leaderIndex] != signer {
+	x.whosTurn = masterNodes[leaderIndex]
-		log.Info("[yourturn] Not my turn", "curIndex", curIndex, "leaderIndex", leaderIndex, "Hash", parent.Hash().Hex(), "masterNodes[leaderIndex]", masterNodes[leaderIndex], "signer", signer)
+	if x.whosTurn != signer {
+		log.Info("[yourturn] Not my turn", "curIndex", curIndex, "leaderIndex", leaderIndex, "Hash", parent.Hash().Hex(), "whosTurn", x.whosTurn, "myaddr", signer)
 		return false, nil
 	}
 

consensus/XDPoS/engines/engine_v2/timeout.go

@@ -189,7 +189,7 @@ func (x *XDPoS_v2) sendTimeout(chain consensus.ChainReader) error {
 		Signature: signedHash,
 		GapNumber: gapNumber,
 	}
-	log.Info("[sendTimeout] Timeout message generated, ready to send!", "timeoutMsgRound", timeoutMsg.Round, "timeoutMsgGapNumber", timeoutMsg.GapNumber)
+	log.Warn("[sendTimeout] Timeout message generated, ready to send!", "timeoutMsgRound", timeoutMsg.Round, "timeoutMsgGapNumber", timeoutMsg.GapNumber, "whosTurn", x.whosTurn)
 	err = x.timeoutHandler(chain, timeoutMsg)
 	if err != nil {
 		log.Error("TimeoutHandler error", "TimeoutRound", timeoutMsg.Round, "Error", err)
@@ -221,7 +221,7 @@ func (x *XDPoS_v2) OnCountdownTimeout(time time.Time, chain interface{}) error {
 
 	x.timeoutCount++
 	if x.timeoutCount%x.config.V2.TimeoutSyncThreshold == 0 {
-		log.Info("[OnCountdownTimeout] timeout sync threadhold reached, send syncInfo message")
+		log.Warn("[OnCountdownTimeout] timeout sync threadhold reached, send syncInfo message")
 		syncInfo := x.getSyncInfo()
 		x.broadcastToBftChannel(syncInfo)
 	}

core/blockchain.go

@@ -2430,6 +2430,11 @@ func (bc *BlockChain) GetHeaderByNumber(number uint64) *types.Header {
 	return bc.hc.GetHeaderByNumber(number)
 }
 
+// Set config for testing purpose function
+func (bc *BlockChain) SetConfig(config *params.ChainConfig) {
+	bc.chainConfig = config
+}
+
 // Config retrieves the blockchain's chain configuration.
 func (bc *BlockChain) Config() *params.ChainConfig { return bc.chainConfig }
 

eth/bft/bft_handler.go

@@ -9,17 +9,25 @@ import (
 	"github.com/XinFinOrg/XDPoSChain/log"
 )
 
+const maxBlockDist = 7 // Maximum allowed backward distance from the chain head, 7 is just a magic number indicate very close block
+
 //Define Boradcast Group functions
 type broadcastVoteFn func(*types.Vote)
 type broadcastTimeoutFn func(*types.Timeout)
 type broadcastSyncInfoFn func(*types.SyncInfo)
 
+// chainHeightFn is a callback type to retrieve the current chain height.
+type chainHeightFn func() uint64
+
 type Bfter struct {
+	gapNumber uint64
+
 	blockChainReader consensus.ChainReader
 	broadcastCh      chan interface{}
 	quit             chan struct{}
 	consensus        ConsensusFns
 	broadcast        BroadcastFns
+	chainHeight      chainHeightFn // Retrieves the current chain's height
 }
 
 type ConsensusFns struct {
@@ -39,16 +47,22 @@ type BroadcastFns struct {
 	SyncInfo broadcastSyncInfoFn
 }
 
-func New(broadcasts BroadcastFns, blockChainReader *core.BlockChain) *Bfter {
+func New(broadcasts BroadcastFns, blockChainReader *core.BlockChain, chainHeight chainHeightFn) *Bfter {
-
 	return &Bfter{
-		quit:             make(chan struct{}),
-		broadcastCh:      make(chan interface{}),
 		broadcast:        broadcasts,
 		blockChainReader: blockChainReader,
+		chainHeight:      chainHeight,
+
+		quit:        make(chan struct{}),
+		broadcastCh: make(chan interface{}),
 	}
 }
 
+// Create this function to avoid massive test change
+func (b *Bfter) InitGapNumber() {
+	b.gapNumber = b.blockChainReader.Config().XDPoS.Gap
+}
+
 func (b *Bfter) SetConsensusFuns(engine consensus.Engine) {
 	e := engine.(*XDPoS.XDPoS)
 	b.broadcastCh = e.EngineV2.BroadcastCh
@@ -63,9 +77,15 @@ func (b *Bfter) SetConsensusFuns(engine consensus.Engine) {
 	}
 }
 
-func (b *Bfter) Vote(vote *types.Vote) error {
+func (b *Bfter) Vote(peer string, vote *types.Vote) error {
 	log.Trace("Receive Vote", "hash", vote.Hash().Hex(), "voted block hash", vote.ProposedBlockInfo.Hash.Hex(), "number", vote.ProposedBlockInfo.Number, "round", vote.ProposedBlockInfo.Round)
 
+	voteBlockNum := vote.ProposedBlockInfo.Number.Int64()
+	if dist := voteBlockNum - int64(b.chainHeight()); dist < -maxBlockDist || dist > maxBlockDist {
+		log.Debug("Discarded propagated vote, too far away", "peer", peer, "number", voteBlockNum, "hash", vote.ProposedBlockInfo.Hash, "distance", dist)
+		return nil
+	}
+
 	verified, err := b.consensus.verifyVote(b.blockChainReader, vote)
 
 	if err != nil {
@@ -89,12 +109,18 @@ func (b *Bfter) Vote(vote *types.Vote) error {
 
 	return nil
 }
-func (b *Bfter) Timeout(timeout *types.Timeout) error {
+func (b *Bfter) Timeout(peer string, timeout *types.Timeout) error {
 	log.Debug("Receive Timeout", "timeout", timeout)
 
+	gapNum := timeout.GapNumber
+	if dist := int64(gapNum) - int64(b.chainHeight()); dist < -int64(b.gapNumber)*2 || dist > int64(b.gapNumber)*2 { // times 2 is to avoid miscalculation on cross epoch case
+		log.Debug("Discarded propagated timeout, too far away", "peer", peer, "gapNumber", gapNum, "hash", timeout.Hash, "distance", dist)
+		return nil
+	}
+
 	verified, err := b.consensus.verifyTimeout(b.blockChainReader, timeout)
 	if err != nil {
-		log.Error("Verify BFT Timeout", "timeoutRound", timeout.Round, "timeoutGapNum", timeout.GapNumber, "error", err)
+		log.Error("Verify BFT Timeout", "timeoutRound", timeout.Round, "timeoutGapNum", gapNum, "error", err)
 		return err
 	}
 
@@ -113,9 +139,15 @@ func (b *Bfter) Timeout(timeout *types.Timeout) error {
 
 	return nil
 }
-func (b *Bfter) SyncInfo(syncInfo *types.SyncInfo) error {
+func (b *Bfter) SyncInfo(peer string, syncInfo *types.SyncInfo) error {
 	log.Debug("Receive SyncInfo", "syncInfo", syncInfo)
 
+	qcBlockNum := syncInfo.HighestQuorumCert.ProposedBlockInfo.Number.Int64()
+	if dist := qcBlockNum - int64(b.chainHeight()); dist < -maxBlockDist || dist > maxBlockDist {
+		log.Debug("Discarded propagated syncInfo, too far away", "peer", peer, "blockNum", qcBlockNum, "hash", syncInfo.Hash, "distance", dist)
+		return nil
+	}
+
 	verified, err := b.consensus.verifySyncInfo(b.blockChainReader, syncInfo)
 	if err != nil {
 		log.Error("Verify BFT SyncInfo", "error", err)

eth/bft/bft_handler_test.go

@@ -2,6 +2,7 @@ package bft
 
 import (
 	"fmt"
+	"math/big"
 	"sync/atomic"
 	"testing"
 	"time"
@@ -12,15 +13,18 @@ import (
 	"github.com/XinFinOrg/XDPoSChain/consensus/XDPoS/utils"
 	"github.com/XinFinOrg/XDPoSChain/core"
 	"github.com/XinFinOrg/XDPoSChain/core/types"
+	"github.com/XinFinOrg/XDPoSChain/params"
 	"github.com/stretchr/testify/assert"
 )
 
+const peerID = "abc"
+
 // make different votes based on Signatures
 func makeVotes(n int) []types.Vote {
 	var votes []types.Vote
 	for i := 0; i < n; i++ {
 		votes = append(votes, types.Vote{
-			ProposedBlockInfo: &types.BlockInfo{},
+			ProposedBlockInfo: &types.BlockInfo{Number: big.NewInt(1)},
 			Signature:         []byte{byte(i)},
 			GapNumber:         0,
 		})
@@ -38,9 +42,14 @@ func newTester() *bfterTester {
 	testConsensus := &XDPoS.XDPoS{EngineV2: &engine_v2.XDPoS_v2{}}
 	broadcasts := BroadcastFns{}
 	blockChain := &core.BlockChain{}
+	blockChain.SetConfig(params.TestXDPoSMockChainConfig)
+	chainHeight := func() uint64 {
+		return 1
+	}
 
 	tester := &bfterTester{}
-	tester.bfter = New(broadcasts, blockChain)
+	tester.bfter = New(broadcasts, blockChain, chainHeight)
+	tester.bfter.InitGapNumber()
 	tester.bfter.SetConsensusFuns(testConsensus)
 	tester.bfter.broadcastCh = make(chan interface{})
 	tester.bfter.Start()
@@ -72,7 +81,7 @@ func TestSequentialVotes(t *testing.T) {
 
 	votes := makeVotes(targetVotes)
 	for _, vote := range votes {
-		err := tester.bfter.Vote(&vote)
+		err := tester.bfter.Vote(peerID, &vote)
 		if err != nil {
 			t.Fatal(err)
 		}
@@ -104,8 +113,8 @@ func TestNotBoardcastInvalidVote(t *testing.T) {
 		atomic.AddUint32(&broadcastCounter, 1)
 	}
 
-	vote := types.Vote{ProposedBlockInfo: &types.BlockInfo{}}
+	vote := types.Vote{ProposedBlockInfo: &types.BlockInfo{Number: big.NewInt(1)}}
-	tester.bfter.Vote(&vote)
+	tester.bfter.Vote(peerID, &vote)
 
 	time.Sleep(50 * time.Millisecond)
 	if int(handlerCounter) != targetVotes || int(broadcastCounter) != targetVotes {
@@ -131,8 +140,8 @@ func TestBoardcastButNotProcessDisqualifiedVotes(t *testing.T) {
 		atomic.AddUint32(&broadcastCounter, 1)
 	}
 
-	vote := types.Vote{ProposedBlockInfo: &types.BlockInfo{}}
+	vote := types.Vote{ProposedBlockInfo: &types.BlockInfo{Number: big.NewInt(1)}}
-	tester.bfter.Vote(&vote)
+	tester.bfter.Vote(peerID, &vote)
 
 	time.Sleep(50 * time.Millisecond)
 	if int(handlerCounter) != targetVotes || int(broadcastCounter) != 1 {
@@ -158,8 +167,8 @@ func TestBoardcastButNotProcessDisqualifiedTimeout(t *testing.T) {
 		atomic.AddUint32(&broadcastCounter, 1)
 	}
 
-	timeout := types.Timeout{}
+	timeout := types.Timeout{GapNumber: 450}
-	tester.bfter.Timeout(&timeout)
+	tester.bfter.Timeout(peerID, &timeout)
 
 	time.Sleep(50 * time.Millisecond)
 	if int(handlerCounter) != targetTimeout || int(broadcastCounter) != 1 {
@@ -185,8 +194,8 @@ func TestBoardcastButNotProcessDisqualifiedSyncInfo(t *testing.T) {
 		atomic.AddUint32(&broadcastCounter, 1)
 	}
 
-	syncInfo := types.SyncInfo{}
+	syncInfo := types.SyncInfo{HighestQuorumCert: &types.QuorumCert{ProposedBlockInfo: &types.BlockInfo{Number: big.NewInt(1)}}}
-	tester.bfter.SyncInfo(&syncInfo)
+	tester.bfter.SyncInfo(peerID, &syncInfo)
 
 	time.Sleep(50 * time.Millisecond)
 	if int(handlerCounter) != targetSyncInfo || int(broadcastCounter) != 1 {
@@ -194,9 +203,6 @@ func TestBoardcastButNotProcessDisqualifiedSyncInfo(t *testing.T) {
 	}
 }
 
-// TODO: SyncInfo and Timeout Test, should be same as Vote.
-// Once all test on vote covered, then duplicate to others
-
 func TestTimeoutHandler(t *testing.T) {
 	tester := newTester()
 	verifyCounter := uint32(0)
@@ -218,9 +224,9 @@ func TestTimeoutHandler(t *testing.T) {
 		atomic.AddUint32(&broadcastCounter, 1)
 	}
 
-	timeoutMsg := &types.Timeout{}
+	timeoutMsg := &types.Timeout{GapNumber: 450}
 
-	err := tester.bfter.Timeout(timeoutMsg)
+	err := tester.bfter.Timeout(peerID, timeoutMsg)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -251,6 +257,137 @@ func TestTimeoutHandlerRoundNotEqual(t *testing.T) {
 
 	timeoutMsg := &types.Timeout{}
 
-	err := tester.bfter.Timeout(timeoutMsg)
+	err := tester.bfter.Timeout(peerID, timeoutMsg)
 	assert.Equal(t, "timeout message round number: 1 does not match currentRound: 2", err.Error())
 }
+
+func TestSyncInfoHandler(t *testing.T) {
+	tester := newTester()
+	verifyCounter := uint32(0)
+	handlerCounter := uint32(0)
+	broadcastCounter := uint32(0)
+	targetSyncInfo := 1
+
+	tester.bfter.consensus.verifySyncInfo = func(chain consensus.ChainReader, syncInfo *types.SyncInfo) (bool, error) {
+		atomic.AddUint32(&verifyCounter, 1)
+		return true, nil // return false but with nil in error means the message is valid but disqualified
+	}
+
+	tester.bfter.consensus.syncInfoHandler = func(chain consensus.ChainReader, syncInfo *types.SyncInfo) error {
+		atomic.AddUint32(&handlerCounter, 1)
+		return nil
+	}
+	tester.bfter.broadcast.SyncInfo = func(*types.SyncInfo) {
+		atomic.AddUint32(&broadcastCounter, 1)
+	}
+
+	syncInfo := types.SyncInfo{HighestQuorumCert: &types.QuorumCert{ProposedBlockInfo: &types.BlockInfo{Number: big.NewInt(1)}}}
+	tester.bfter.SyncInfo(peerID, &syncInfo)
+
+	time.Sleep(50 * time.Millisecond)
+	if int(verifyCounter) != targetSyncInfo || int(handlerCounter) != targetSyncInfo || int(broadcastCounter) != 1 {
+		t.Fatalf("count mismatch: have %v on verify, have %v on handler, %v on broadcast, want %v", verifyCounter, handlerCounter, broadcastCounter, targetSyncInfo)
+	}
+}
+
+func TestTooFarVotes(t *testing.T) {
+	tester := newTester()
+	verifyCounter := uint32(0)
+	handlerCounter := uint32(0)
+	broadcastCounter := uint32(0)
+	numberVotes := 10
+	targetVotes := 0
+
+	tester.bfter.consensus.verifyVote = func(chain consensus.ChainReader, vote *types.Vote) (bool, error) {
+		atomic.AddUint32(&verifyCounter, 1)
+		return true, nil
+	}
+
+	tester.bfter.consensus.voteHandler = func(chain consensus.ChainReader, vote *types.Vote) error {
+		atomic.AddUint32(&handlerCounter, 1)
+		return nil
+	}
+
+	tester.bfter.broadcast.Vote = func(*types.Vote) {
+		atomic.AddUint32(&broadcastCounter, 1)
+	}
+
+	tester.bfter.chainHeight = func() uint64 { return 100 }
+
+	votes := makeVotes(numberVotes)
+	for _, vote := range votes {
+		err := tester.bfter.Vote(peerID, &vote)
+		if err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	time.Sleep(100 * time.Millisecond)
+	if int(verifyCounter) != targetVotes || int(handlerCounter) != targetVotes || int(broadcastCounter) != targetVotes {
+		t.Fatalf("count mismatch: have %v on verify, %v on handler, %v on broadcast, want %v", verifyCounter, handlerCounter, broadcastCounter, targetVotes)
+	}
+}
+
+func TestTooFarTimeout(t *testing.T) {
+	tester := newTester()
+	verifyCounter := uint32(0)
+	handlerCounter := uint32(0)
+	broadcastCounter := uint32(0)
+	targetTimeout := 0
+
+	tester.bfter.consensus.verifyTimeout = func(consensus.ChainReader, *types.Timeout) (bool, error) {
+		atomic.AddUint32(&verifyCounter, 1)
+		return true, nil
+	}
+
+	tester.bfter.consensus.timeoutHandler = func(chain consensus.ChainReader, timeout *types.Timeout) error {
+		atomic.AddUint32(&handlerCounter, 1)
+		return nil
+	}
+
+	tester.bfter.broadcast.Timeout = func(*types.Timeout) {
+		atomic.AddUint32(&broadcastCounter, 1)
+	}
+
+	timeoutMsg := &types.Timeout{GapNumber: 10000}
+
+	err := tester.bfter.Timeout(peerID, timeoutMsg)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	time.Sleep(100 * time.Millisecond)
+
+	if int(verifyCounter) != targetTimeout || int(handlerCounter) != targetTimeout || int(broadcastCounter) != targetTimeout {
+		t.Fatalf("count mismatch: have %v on verify, %v on handler, %v on broadcast, want %v", verifyCounter, handlerCounter, broadcastCounter, targetTimeout)
+	}
+}
+
+func TestTooFarSyncInfo(t *testing.T) {
+	tester := newTester()
+	verifyCounter := uint32(0)
+	handlerCounter := uint32(0)
+	broadcastCounter := uint32(0)
+	targetSyncInfo := 0
+
+	tester.bfter.consensus.verifySyncInfo = func(chain consensus.ChainReader, syncInfo *types.SyncInfo) (bool, error) {
+		atomic.AddUint32(&verifyCounter, 1)
+		return true, nil // return false but with nil in error means the message is valid but disqualified
+	}
+
+	tester.bfter.consensus.syncInfoHandler = func(chain consensus.ChainReader, syncInfo *types.SyncInfo) error {
+		atomic.AddUint32(&handlerCounter, 1)
+		return nil
+	}
+	tester.bfter.broadcast.SyncInfo = func(*types.SyncInfo) {
+		atomic.AddUint32(&broadcastCounter, 1)
+	}
+
+	syncInfo := types.SyncInfo{HighestQuorumCert: &types.QuorumCert{ProposedBlockInfo: &types.BlockInfo{Number: big.NewInt(100)}}}
+	tester.bfter.SyncInfo(peerID, &syncInfo)
+
+	time.Sleep(50 * time.Millisecond)
+	if int(verifyCounter) != targetSyncInfo || int(handlerCounter) != targetSyncInfo || int(broadcastCounter) != targetSyncInfo {
+		t.Fatalf("count mismatch: have %v on verify, have %v on handler, %v on broadcast, want %v", verifyCounter, handlerCounter, broadcastCounter, targetSyncInfo)
+	}
+}

eth/handler.go

@@ -227,6 +227,7 @@ func NewProtocolManager(config *params.ChainConfig, mode downloader.SyncMode, ne
 	heighter := func() uint64 {
 		return blockchain.CurrentBlock().NumberU64()
 	}
+
 	inserter := func(block *types.Block) error {
 		// If fast sync is running, deny importing weird blocks
 		if atomic.LoadUint32(&manager.fastSync) == 1 {
@@ -253,8 +254,9 @@ func NewProtocolManager(config *params.ChainConfig, mode downloader.SyncMode, ne
 		Timeout:  manager.BroadcastTimeout,
 		SyncInfo: manager.BroadcastSyncInfo,
 	}
-	manager.bft = bft.New(broadcasts, blockchain)
+	manager.bft = bft.New(broadcasts, blockchain, heighter)
 	if blockchain.Config().XDPoS != nil {
+		manager.bft.InitGapNumber()
 		manager.bft.SetConsensusFuns(engine)
 	}
 
@@ -860,7 +862,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
 
 		exist, _ := pm.knownVotes.ContainsOrAdd(vote.Hash(), true)
 		if !exist {
-			go pm.bft.Vote(&vote)
+			go pm.bft.Vote(p.id, &vote)
 		} else {
 			log.Debug("Discarded vote, known vote", "vote hash", vote.Hash(), "voted block hash", vote.ProposedBlockInfo.Hash.Hex(), "number", vote.ProposedBlockInfo.Number, "round", vote.ProposedBlockInfo.Round)
 		}
@@ -880,7 +882,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
 		exist, _ := pm.knownTimeouts.ContainsOrAdd(timeout.Hash(), true)
 
 		if !exist {
-			go pm.bft.Timeout(&timeout)
+			go pm.bft.Timeout(p.id, &timeout)
 		} else {
 			log.Trace("Discarded Timeout, known Timeout", "Signature", timeout.Signature, "hash", timeout.Hash(), "round", timeout.Round)
 		}
@@ -899,7 +901,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
 
 		exist, _ := pm.knownSyncInfos.ContainsOrAdd(syncInfo.Hash(), true)
 		if !exist {
-			go pm.bft.SyncInfo(&syncInfo)
+			go pm.bft.SyncInfo(p.id, &syncInfo)
 		} else {
 			log.Trace("Discarded SyncInfo, known SyncInfo", "hash", syncInfo.Hash())
 		}