Timeout for block proposal. (#4553)

* Timeout for proposal.

cmd/harmony/main.go

@@ -14,28 +14,12 @@ import (
 	"syscall"
 	"time"
 
-	"github.com/harmony-one/harmony/consensus/quorum"
-	"github.com/harmony-one/harmony/internal/chain"
-	"github.com/harmony-one/harmony/internal/registry"
-	"github.com/harmony-one/harmony/internal/shardchain/tikv_manage"
-	"github.com/harmony-one/harmony/internal/tikv/redis_helper"
-	"github.com/harmony-one/harmony/internal/tikv/statedb_cache"
-
-	"github.com/harmony-one/harmony/api/service/crosslink_sending"
-	rosetta_common "github.com/harmony-one/harmony/rosetta/common"
-
-	harmonyconfig "github.com/harmony-one/harmony/internal/configs/harmony"
-	rpc_common "github.com/harmony-one/harmony/rpc/common"
-
 	ethCommon "github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/common/hexutil"
 	"github.com/ethereum/go-ethereum/log"
-	"github.com/pkg/errors"
-	"github.com/spf13/cobra"
-
 	"github.com/harmony-one/bls/ffi/go/bls"
-
 	"github.com/harmony-one/harmony/api/service"
+	"github.com/harmony-one/harmony/api/service/crosslink_sending"
 	"github.com/harmony-one/harmony/api/service/pprof"
 	"github.com/harmony-one/harmony/api/service/prometheus"
 	"github.com/harmony-one/harmony/api/service/stagedstreamsync"
@@ -43,22 +27,33 @@ import (
 	"github.com/harmony-one/harmony/common/fdlimit"
 	"github.com/harmony-one/harmony/common/ntp"
 	"github.com/harmony-one/harmony/consensus"
+	"github.com/harmony-one/harmony/consensus/quorum"
 	"github.com/harmony-one/harmony/core"
 	"github.com/harmony-one/harmony/hmy/downloader"
+	"github.com/harmony-one/harmony/internal/chain"
 	"github.com/harmony-one/harmony/internal/cli"
 	"github.com/harmony-one/harmony/internal/common"
+	harmonyconfig "github.com/harmony-one/harmony/internal/configs/harmony"
 	nodeconfig "github.com/harmony-one/harmony/internal/configs/node"
 	shardingconfig "github.com/harmony-one/harmony/internal/configs/sharding"
 	"github.com/harmony-one/harmony/internal/genesis"
 	"github.com/harmony-one/harmony/internal/params"
+	"github.com/harmony-one/harmony/internal/registry"
 	"github.com/harmony-one/harmony/internal/shardchain"
+	"github.com/harmony-one/harmony/internal/shardchain/tikv_manage"
+	"github.com/harmony-one/harmony/internal/tikv/redis_helper"
+	"github.com/harmony-one/harmony/internal/tikv/statedb_cache"
 	"github.com/harmony-one/harmony/internal/utils"
 	"github.com/harmony-one/harmony/multibls"
 	"github.com/harmony-one/harmony/node"
 	"github.com/harmony-one/harmony/numeric"
 	"github.com/harmony-one/harmony/p2p"
+	rosetta_common "github.com/harmony-one/harmony/rosetta/common"
+	rpc_common "github.com/harmony-one/harmony/rpc/common"
 	"github.com/harmony-one/harmony/shard"
 	"github.com/harmony-one/harmony/webhooks"
+	"github.com/pkg/errors"
+	"github.com/spf13/cobra"
 )
 
 // Host

consensus/consensus_test.go

@@ -64,9 +64,6 @@ func TestConsensusInitialization(t *testing.T) {
 	assert.IsType(t, make(chan slash.Record), consensus.SlashChan)
 	assert.NotNil(t, consensus.SlashChan)
 
-	assert.IsType(t, make(chan ProposalType), consensus.GetReadySignal())
-	assert.NotNil(t, consensus.GetReadySignal())
-
 	assert.IsType(t, make(chan [vdfAndSeedSize]byte), consensus.RndChannel)
 	assert.NotNil(t, consensus.RndChannel)
 

consensus/consensus_v2.go

@@ -10,24 +10,23 @@ import (
 
 	"github.com/ethereum/go-ethereum/common"
 	bls2 "github.com/harmony-one/bls/ffi/go/bls"
-	"github.com/harmony-one/harmony/consensus/signature"
-	"github.com/harmony-one/harmony/core"
-	nodeconfig "github.com/harmony-one/harmony/internal/configs/node"
-	"github.com/harmony-one/harmony/internal/utils"
-	libp2p_peer "github.com/libp2p/go-libp2p/core/peer"
-	"github.com/rs/zerolog"
-
 	msg_pb "github.com/harmony-one/harmony/api/proto/message"
 	"github.com/harmony-one/harmony/block"
 	"github.com/harmony-one/harmony/consensus/quorum"
+	"github.com/harmony-one/harmony/consensus/signature"
+	"github.com/harmony-one/harmony/core"
 	"github.com/harmony-one/harmony/core/types"
 	"github.com/harmony-one/harmony/crypto/bls"
 	vrf_bls "github.com/harmony-one/harmony/crypto/vrf/bls"
+	nodeconfig "github.com/harmony-one/harmony/internal/configs/node"
+	"github.com/harmony-one/harmony/internal/utils"
 	"github.com/harmony-one/harmony/p2p"
 	"github.com/harmony-one/harmony/shard"
 	"github.com/harmony-one/vdf/src/vdf_go"
+	libp2p_peer "github.com/libp2p/go-libp2p/core/peer"
 	"github.com/pkg/errors"
 	"github.com/prometheus/client_golang/prometheus"
+	"github.com/rs/zerolog"
 )
 
 var (
@@ -681,37 +680,36 @@ func (consensus *Consensus) commitBlock(blk *types.Block, committedMsg *FBFTMess
 
 // rotateLeader rotates the leader to the next leader in the committee.
 // This function must be called with enabled leader rotation.
-func (consensus *Consensus) rotateLeader(epoch *big.Int) {
+func (consensus *Consensus) rotateLeader(epoch *big.Int) *bls.PublicKeyWrapper {
 	var (
 		bc        = consensus.Blockchain()
-		prev      = consensus.getLeaderPubKey()
 		leader    = consensus.getLeaderPubKey()
 		curBlock  = bc.CurrentBlock()
 		curNumber = curBlock.NumberU64()
 		curEpoch  = curBlock.Epoch().Uint64()
 	)
-	const blocksCountAliveness = 10
+	const blocksCountAliveness = 4
 
 	utils.Logger().Info().Msgf("[Rotating leader] epoch: %v rotation:%v external rotation %v", epoch.Uint64(), bc.Config().IsLeaderRotationInternalValidators(epoch), bc.Config().IsLeaderRotationExternalValidatorsAllowed(epoch))
 	ss, err := bc.ReadShardState(epoch)
 	if err != nil {
 		utils.Logger().Error().Err(err).Msg("Failed to read shard state")
-		return
+		return nil
 	}
 	committee, err := ss.FindCommitteeByID(consensus.ShardID)
 	if err != nil {
 		utils.Logger().Error().Err(err).Msg("Failed to find committee")
-		return
+		return nil
 	}
 	slotsCount := len(committee.Slots)
 	blocksPerEpoch := shard.Schedule.InstanceForEpoch(epoch).BlocksPerEpoch()
 	if blocksPerEpoch == 0 {
 		utils.Logger().Error().Msg("[Rotating leader] blocks per epoch is 0")
-		return
+		return nil
 	}
 	if slotsCount == 0 {
 		utils.Logger().Error().Msg("[Rotating leader] slots count is 0")
-		return
+		return nil
 	}
 	numBlocksProducedByLeader := blocksPerEpoch / uint64(slotsCount)
 	rest := blocksPerEpoch % uint64(slotsCount)
@@ -723,7 +721,7 @@ func (consensus *Consensus) rotateLeader(epoch *big.Int) {
 	s := bc.LeaderRotationMeta()
 	if !bytes.Equal(leader.Bytes[:], s.Pub) {
 		// Another leader.
-		return
+		return nil
 	}
 	// If it is the first validator producing blocks, it should also produce the remaining 'rest' of the blocks.
 	if s.Shifts == 0 {
@@ -731,7 +729,7 @@ func (consensus *Consensus) rotateLeader(epoch *big.Int) {
 	}
 	if s.Count < numBlocksProducedByLeader {
 		// Not enough blocks produced by the leader, continue producing by the same leader.
-		return
+		return nil
 	}
 	// Passed all checks, we can change leader.
 	// NthNext will move the leader to the next leader in the committee.
@@ -742,7 +740,7 @@ func (consensus *Consensus) rotateLeader(epoch *big.Int) {
 		offset   = 1
 	)
 
-	for {
+	for i := 0; i < len(committee.Slots); i++ {
 		if bc.Config().IsLeaderRotationExternalValidatorsAllowed(epoch) {
 			wasFound, next = consensus.Decider.NthNextValidator(committee.Slots, leader, offset)
 		} else {
@@ -751,7 +749,7 @@ func (consensus *Consensus) rotateLeader(epoch *big.Int) {
 		if !wasFound {
 			utils.Logger().Error().Msg("Failed to get next leader")
 			// Seems like nothing we can do here.
-			return
+			return nil
 		}
 		members := consensus.Decider.Participants()
 		mask := bls.NewMask(members)
@@ -760,7 +758,7 @@ func (consensus *Consensus) rotateLeader(epoch *big.Int) {
 			header := bc.GetHeaderByNumber(curNumber - uint64(i))
 			if header == nil {
 				utils.Logger().Error().Msgf("Failed to get header by number %d", curNumber-uint64(i))
-				return
+				return nil
 			}
 			// if epoch is different, we should not check this block.
 			if header.Epoch().Uint64() != curEpoch {
@@ -770,12 +768,12 @@ func (consensus *Consensus) rotateLeader(epoch *big.Int) {
 			err = mask.SetMask(header.LastCommitBitmap())
 			if err != nil {
 				utils.Logger().Err(err).Msg("Failed to set mask")
-				return
+				return nil
 			}
 			ok, err := mask.KeyEnabled(next.Bytes)
 			if err != nil {
 				utils.Logger().Err(err).Msg("Failed to get key enabled")
-				return
+				return nil
 			}
 			if !ok {
 				skipped++
@@ -788,16 +786,9 @@ func (consensus *Consensus) rotateLeader(epoch *big.Int) {
 			offset++
 			continue
 		}
-		consensus.setLeaderPubKey(next)
-		break
-	}
-
-	if consensus.isLeader() && !consensus.getLeaderPubKey().Object.IsEqual(prev.Object) {
-		// leader changed
-		go func() {
-			consensus.ReadySignal(SyncProposal)
-		}()
+		return next
 	}
+	return nil
 }
 
 // SetupForNewConsensus sets the state for new consensus
@@ -812,7 +803,18 @@ func (consensus *Consensus) setupForNewConsensus(blk *types.Block, committedMsg
 		epoch = blk.Epoch()
 	}
 	if consensus.Blockchain().Config().IsLeaderRotationInternalValidators(epoch) {
-		consensus.rotateLeader(epoch)
+		if next := consensus.rotateLeader(epoch); next != nil {
+			prev := consensus.getLeaderPubKey()
+			consensus.setLeaderPubKey(next)
+			if consensus.isLeader() && !consensus.getLeaderPubKey().Object.IsEqual(prev.Object) {
+				// leader changed
+				blockPeriod := consensus.BlockPeriod
+				go func() {
+					<-time.After(blockPeriod)
+					consensus.ReadySignal(SyncProposal)
+				}()
+			}
+		}
 	}
 
 	// Update consensus keys at last so the change of leader status doesn't mess up normal flow

consensus/fbft_log.go

@@ -3,6 +3,8 @@ package consensus
 import (
 	"encoding/binary"
 	"fmt"
+	"hash/crc32"
+	"strconv"
 	"sync"
 
 	"github.com/ethereum/go-ethereum/common"
@@ -36,6 +38,19 @@ type FBFTMessage struct {
 	Verified           bool
 }
 
+func (m *FBFTMessage) Hash() []byte {
+	// Hash returns hash of the struct
+
+	c := crc32.NewIEEE()
+	c.Write([]byte(strconv.FormatUint(uint64(m.MessageType), 10)))
+	c.Write([]byte(strconv.FormatUint(m.ViewID, 10)))
+	c.Write([]byte(strconv.FormatUint(m.BlockNum, 10)))
+	c.Write(m.BlockHash[:])
+	c.Write(m.Block[:])
+	c.Write(m.Payload[:])
+	return c.Sum(nil)
+}
+
 // String ..
 func (m *FBFTMessage) String() string {
 	sender := ""

consensus/leader.go

@@ -7,6 +7,7 @@ import (
 	"github.com/harmony-one/harmony/crypto/bls"
 	"github.com/harmony-one/harmony/internal/common"
 	nodeconfig "github.com/harmony-one/harmony/internal/configs/node"
+	"github.com/harmony-one/harmony/internal/utils/rclient"
 
 	"github.com/ethereum/go-ethereum/rlp"
 	bls_core "github.com/harmony-one/bls/ffi/go/bls"
@@ -16,6 +17,7 @@ import (
 	"github.com/harmony-one/harmony/p2p"
 )
 
+// announce fires leader
 func (consensus *Consensus) announce(block *types.Block) {
 	blockHash := block.Hash()
 
@@ -92,6 +94,7 @@ func (consensus *Consensus) announce(block *types.Block) {
 	consensus.switchPhase("Announce", FBFTPrepare)
 }
 
+// this method is called for each validator sent their vote message
 func (consensus *Consensus) onPrepare(recvMsg *FBFTMessage) {
 	// TODO(audit): make FBFT lookup using map instead of looping through all items.
 	if !consensus.fBFTLog.HasMatchingViewAnnounce(
@@ -189,6 +192,7 @@ func (consensus *Consensus) onPrepare(recvMsg *FBFTMessage) {
 	//// Read - End
 }
 
+// this method is called by leader
 func (consensus *Consensus) onCommit(recvMsg *FBFTMessage) {
 	//// Read - Start
 	if !consensus.isRightBlockNumAndViewID(recvMsg) {

go.mod

@@ -68,10 +68,12 @@ require (
 
 require (
 	github.com/c2h5oh/datasize v0.0.0-20220606134207-859f65c6625b
+	github.com/grafana/pyroscope-go v1.0.4
 	github.com/holiman/bloomfilter/v2 v2.0.3
 	github.com/ledgerwatch/erigon-lib v0.0.0-20230607152933-42c9c28cac68
 	github.com/ledgerwatch/log/v3 v3.8.0
 	github.com/olekukonko/tablewriter v0.0.5
+	golang.org/x/exp v0.0.0-20231006140011-7918f672742d
 )
 
 require (
@@ -146,6 +148,7 @@ require (
 	github.com/google/pprof v0.0.0-20230405160723-4a4c7d95572b // indirect
 	github.com/google/uuid v1.3.0 // indirect
 	github.com/gostaticanalysis/analysisutil v0.0.0-20190318220348-4088753ea4d3 // indirect
+	github.com/grafana/pyroscope-go/godeltaprof v0.1.4 // indirect
 	github.com/grpc-ecosystem/go-grpc-middleware v1.4.0 // indirect
 	github.com/hashicorp/errwrap v1.1.0 // indirect
 	github.com/hashicorp/go-multierror v1.1.1 // indirect
@@ -262,7 +265,6 @@ require (
 	go.uber.org/dig v1.16.1 // indirect
 	go.uber.org/fx v1.19.2 // indirect
 	go.uber.org/multierr v1.11.0 // indirect
-	golang.org/x/exp v0.0.0-20231006140011-7918f672742d // indirect
 	golang.org/x/mod v0.13.0 // indirect
 	golang.org/x/term v0.13.0 // indirect
 	golang.org/x/text v0.13.0 // indirect

go.sum

@@ -632,6 +632,10 @@ github.com/gorilla/websocket v1.5.0 h1:PPwGk2jz7EePpoHN/+ClbZu8SPxiqlu12wZP/3sWm
 github.com/gorilla/websocket v1.5.0/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
 github.com/gostaticanalysis/analysisutil v0.0.0-20190318220348-4088753ea4d3 h1:JVnpOZS+qxli+rgVl98ILOXVNbW+kb5wcxeGx8ShUIw=
 github.com/gostaticanalysis/analysisutil v0.0.0-20190318220348-4088753ea4d3/go.mod h1:eEOZF4jCKGi+aprrirO9e7WKB3beBRtWgqGunKl6pKE=
+github.com/grafana/pyroscope-go v1.0.4 h1:oyQX0BOkL+iARXzHuCdIF5TQ7/sRSel1YFViMHC7Bm0=
+github.com/grafana/pyroscope-go v1.0.4/go.mod h1:0d7ftwSMBV/Awm7CCiYmHQEG8Y44Ma3YSjt+nWcWztY=
+github.com/grafana/pyroscope-go/godeltaprof v0.1.4 h1:mDsJ3ngul7UfrHibGQpV66PbZ3q1T8glz/tK3bQKKEk=
+github.com/grafana/pyroscope-go/godeltaprof v0.1.4/go.mod h1:1HSPtjU8vLG0jE9JrTdzjgFqdJ/VgN7fvxBNq3luJko=
 github.com/graph-gophers/graphql-go v1.3.0/go.mod h1:9CQHMSxwO4MprSdzoIEobiHpoLtHm77vfxsvsIN5Vuc=
 github.com/gregjones/httpcache v0.0.0-20180305231024-9cad4c3443a7/go.mod h1:FecbI9+v66THATjSRHfNgh1IVFe/9kFxbXtjV0ctIMA=
 github.com/grpc-ecosystem/go-grpc-middleware v1.0.0/go.mod h1:FiyG127CGDf3tlThmgyCl78X/SZQqEOJBCDaAfeWzPs=