diff --git a/consensus/XDPoS/engines/engine_v2/engine.go b/consensus/XDPoS/engines/engine_v2/engine.go index 3699cea8ec..530233349d 100644 --- a/consensus/XDPoS/engines/engine_v2/engine.go +++ b/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 } diff --git a/consensus/XDPoS/engines/engine_v2/mining.go b/consensus/XDPoS/engines/engine_v2/mining.go index 951d0eaa14..b12e430d93 100644 --- a/consensus/XDPoS/engines/engine_v2/mining.go +++ b/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 { - log.Info("[yourturn] Not my turn", "curIndex", curIndex, "leaderIndex", leaderIndex, "Hash", parent.Hash().Hex(), "masterNodes[leaderIndex]", masterNodes[leaderIndex], "signer", signer) + x.whosTurn = masterNodes[leaderIndex] + 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 } diff --git a/consensus/XDPoS/engines/engine_v2/timeout.go b/consensus/XDPoS/engines/engine_v2/timeout.go index 9dacc773df..9ab042c91c 100644 --- a/consensus/XDPoS/engines/engine_v2/timeout.go +++ b/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) } diff --git a/core/blockchain.go b/core/blockchain.go index c9f83bf72b..9b23f71333 100644 --- a/core/blockchain.go +++ b/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 } diff --git a/eth/bft/bft_handler.go b/eth/bft/bft_handler.go index c4271b4b22..27b0b92989 100644 --- a/eth/bft/bft_handler.go +++ b/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) diff --git a/eth/bft/bft_handler_test.go b/eth/bft/bft_handler_test.go index c74d0a2bb3..d68bfb5333 100644 --- a/eth/bft/bft_handler_test.go +++ b/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{}} - tester.bfter.Vote(&vote) + vote := types.Vote{ProposedBlockInfo: &types.BlockInfo{Number: big.NewInt(1)}} + 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{}} - tester.bfter.Vote(&vote) + vote := types.Vote{ProposedBlockInfo: &types.BlockInfo{Number: big.NewInt(1)}} + 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{} - tester.bfter.Timeout(&timeout) + timeout := types.Timeout{GapNumber: 450} + 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{} - tester.bfter.SyncInfo(&syncInfo) + 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(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) + } +} diff --git a/eth/handler.go b/eth/handler.go index 6f021a0ac3..8b89a984d8 100644 --- a/eth/handler.go +++ b/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()) }