Merge pull request #3419 from rlan35/fix_vc_commit

Fix view change stuck issue

api/service/syncing/syncing.go

@@ -32,7 +32,7 @@ const (
 	TimesToFail                     = 5 // downloadBlocks service retry limit
 	RegistrationNumber              = 3
 	SyncingPortDifference           = 3000
-	inSyncThreshold                 = 0    // when peerBlockHeight - myBlockHeight <= inSyncThreshold, it's ready to join consensus
+	inSyncThreshold                 = 1    // when peerBlockHeight - myBlockHeight <= inSyncThreshold, it's ready to join consensus
 	SyncLoopBatchSize        uint32 = 1000 // maximum size for one query of block hashes
 	verifyHeaderBatchSize    uint64 = 100  // block chain header verification batch size
 	SyncLoopFrequency               = 1    // unit in second

consensus/consensus_service.go

@@ -421,6 +421,12 @@ func (consensus *Consensus) UpdateConsensusInformation() Mode {
 		Msg("[UpdateConsensusInformation] changing committee")
 
 	// take care of possible leader change during the epoch
+	// TODO: in a very rare case, when a M1 view change happened, the block contains coinbase for last leader
+	// but the new leader is actually recognized by most of the nodes. At this time, if a node sync to this
+	// exact block and set its leader, it will set with the failed leader as in the coinbase of the block.
+	// This is a very rare case scenario and not likely to cause any issue in mainnet. But we need to think about
+	// a solution to take care of this case because the coinbase of the latest block doesn't really represent the
+	// the real current leader in case of M1 view change.
 	if !curHeader.IsLastBlockInEpoch() && curHeader.Number().Uint64() != 0 {
 		leaderPubKey, err := consensus.getLeaderPubKeyFromCoinbase(curHeader)
 		if err != nil || leaderPubKey == nil {
@@ -573,9 +579,9 @@ func (consensus *Consensus) selfCommit(payload []byte) error {
 			continue
 		}
 
-		if _, err := consensus.Decider.SubmitVote(
+		if _, err := consensus.Decider.AddNewVote(
 			quorum.Commit,
-			[]bls.SerializedPublicKey{key.Pub.Bytes},
+			[]*bls_cosi.PublicKeyWrapper{key.Pub},
 			key.Pri.SignHash(commitPayload),
 			common.BytesToHash(consensus.blockHash[:]),
 			block.NumberU64(),

consensus/consensus_v2.go

@@ -517,7 +517,7 @@ func (consensus *Consensus) commitBlock(blk *types.Block, committedMsg *FBFTMess
 	}
 
 	atomic.AddUint64(&consensus.blockNum, 1)
-	consensus.SetCurBlockViewID(committedMsg.ViewID + 1)
+	consensus.SetViewIDs(committedMsg.ViewID + 1)
 	consensus.LeaderPubKey = committedMsg.SenderPubkeys[0]
 	// Update consensus keys at last so the change of leader status doesn't mess up normal flow
 	if blk.IsLastBlockInEpoch() {

consensus/construct_test.go

@@ -71,19 +71,21 @@ func TestConstructPreparedMessage(test *testing.T) {
 	message := "test string"
 	leaderKey := bls.SerializedPublicKey{}
 	leaderKey.FromLibBLSPublicKey(leaderPubKey)
+	leaderKeyWrapper := bls.PublicKeyWrapper{Object: leaderPubKey, Bytes: leaderKey}
 	validatorKey := bls.SerializedPublicKey{}
 	validatorKey.FromLibBLSPublicKey(validatorPubKey)
-	consensus.Decider.SubmitVote(
+	validatorKeyWrapper := bls.PublicKeyWrapper{Object: validatorPubKey, Bytes: validatorKey}
+	consensus.Decider.AddNewVote(
 		quorum.Prepare,
-		[]bls.SerializedPublicKey{leaderKey},
+		[]*bls.PublicKeyWrapper{&leaderKeyWrapper},
 		leaderPriKey.Sign(message),
 		common.BytesToHash(consensus.blockHash[:]),
 		consensus.blockNum,
 		consensus.GetCurBlockViewID(),
 	)
-	if _, err := consensus.Decider.SubmitVote(
+	if _, err := consensus.Decider.AddNewVote(
 		quorum.Prepare,
-		[]bls.SerializedPublicKey{validatorKey},
+		[]*bls.PublicKeyWrapper{&validatorKeyWrapper},
 		validatorPriKey.Sign(message),
 		common.BytesToHash(consensus.blockHash[:]),
 		consensus.blockNum,

consensus/quorum/one-node-one-vote.go

@@ -36,7 +36,7 @@ func (v *uniformVoteWeight) AddNewVote(
 	for i, pubKey := range pubKeys {
 		pubKeysBytes[i] = pubKey.Bytes
 	}
-	return v.SubmitVote(p, pubKeysBytes, sig, headerHash, height, viewID)
+	return v.submitVote(p, pubKeysBytes, sig, headerHash, height, viewID)
 }
 
 // IsQuorumAchieved ..

consensus/quorum/one-node-staked-vote.go

@@ -82,7 +82,7 @@ func (v *stakedVoteWeight) AddNewVote(
 		pubKeysBytes[i] = pubKey.Bytes
 	}
 
-	ballet, err := v.SubmitVote(p, pubKeysBytes, sig, headerHash, height, viewID)
+	ballet, err := v.submitVote(p, pubKeysBytes, sig, headerHash, height, viewID)
 
 	if err != nil {
 		return ballet, err

consensus/quorum/quorom_test.go

@@ -88,7 +88,7 @@ func TestSubmitVote(test *testing.T) {
 
 	decider.UpdateParticipants([]bls.PublicKeyWrapper{pubKeyWrapper1, pubKeyWrapper2})
 
-	if _, err := decider.SubmitVote(
+	if _, err := decider.submitVote(
 		Prepare,
 		[]bls.SerializedPublicKey{pubKeyWrapper1.Bytes},
 		blsPriKey1.Sign(message),
@@ -99,7 +99,7 @@ func TestSubmitVote(test *testing.T) {
 		test.Log(err)
 	}
 
-	if _, err := decider.SubmitVote(
+	if _, err := decider.submitVote(
 		Prepare,
 		[]bls.SerializedPublicKey{pubKeyWrapper2.Bytes},
 		blsPriKey2.Sign(message),
@@ -110,7 +110,7 @@ func TestSubmitVote(test *testing.T) {
 		test.Log(err)
 	}
 	if decider.SignersCount(Prepare) != 2 {
-		test.Fatal("SubmitVote failed")
+		test.Fatal("submitVote failed")
 	}
 
 	aggSig := &bls_core.Sign{}
@@ -145,7 +145,7 @@ func TestSubmitVoteAggregateSig(test *testing.T) {
 
 	decider.UpdateParticipants([]bls.PublicKeyWrapper{pubKeyWrapper1, pubKeyWrapper2})
 
-	decider.SubmitVote(
+	decider.submitVote(
 		Prepare,
 		[]bls.SerializedPublicKey{pubKeyWrapper1.Bytes},
 		blsPriKey1.SignHash(blockHash[:]),
@@ -160,7 +160,7 @@ func TestSubmitVoteAggregateSig(test *testing.T) {
 			aggSig.Add(s)
 		}
 	}
-	if _, err := decider.SubmitVote(
+	if _, err := decider.submitVote(
 		Prepare,
 		[]bls.SerializedPublicKey{pubKeyWrapper2.Bytes, pubKeyWrapper3.Bytes},
 		aggSig,
@@ -172,7 +172,7 @@ func TestSubmitVoteAggregateSig(test *testing.T) {
 	}
 
 	if decider.SignersCount(Prepare) != 3 {
-		test.Fatal("SubmitVote failed")
+		test.Fatal("submitVote failed")
 	}
 
 	aggSig.Add(blsPriKey1.SignHash(blockHash[:]))
@@ -180,7 +180,7 @@ func TestSubmitVoteAggregateSig(test *testing.T) {
 		test.Fatal("AggregateVotes failed")
 	}
 
-	if _, err := decider.SubmitVote(
+	if _, err := decider.submitVote(
 		Prepare,
 		[]bls.SerializedPublicKey{pubKeyWrapper2.Bytes},
 		aggSig,

consensus/quorum/quorum.go

@@ -81,7 +81,8 @@ type ParticipantTracker interface {
 // SignatoryTracker ..
 type SignatoryTracker interface {
 	ParticipantTracker
-	SubmitVote(
+	// This func shouldn't be called directly from outside of quorum. Use AddNewVote instead.
+	submitVote(
 		p Phase, pubkeys []bls.SerializedPublicKey,
 		sig *bls_core.Sign, headerHash common.Hash,
 		height, viewID uint64,
@@ -118,6 +119,7 @@ type Decider interface {
 	DependencyInjectionWriter
 	SetVoters(subCommittee *shard.Committee, epoch *big.Int) (*TallyResult, error)
 	Policy() Policy
+	// Add new vote will add the signature in the memory and increase the cumulative voting power
 	AddNewVote(
 		p Phase, pubkeys []*bls_cosi.PublicKeyWrapper,
 		sig *bls_core.Sign, headerHash common.Hash,
@@ -273,7 +275,7 @@ func (s *cIdentities) SignersCount(p Phase) int64 {
 	}
 }
 
-func (s *cIdentities) SubmitVote(
+func (s *cIdentities) submitVote(
 	p Phase, pubkeys []bls.SerializedPublicKey,
 	sig *bls_core.Sign, headerHash common.Hash,
 	height, viewID uint64,

consensus/validator.go

@@ -114,10 +114,6 @@ func (consensus *Consensus) onPrepared(msg *msg_pb.Message) {
 			Msg("Wrong BlockNum Received, ignoring!")
 		return
 	}
-	if recvMsg.BlockNum > consensus.blockNum {
-		consensus.getLogger().Warn().Msgf("[OnPrepared] low consensus block number. Spin sync")
-		consensus.spinUpStateSync()
-	}
 
 	// check validity of prepared signature
 	blockHash := recvMsg.BlockHash
@@ -153,6 +149,12 @@ func (consensus *Consensus) onPrepared(msg *msg_pb.Message) {
 	if !consensus.onPreparedSanityChecks(&blockObj, recvMsg) {
 		return
 	}
+
+	if recvMsg.BlockNum > consensus.blockNum {
+		consensus.getLogger().Warn().Msgf("[OnPrepared] low consensus block number. Spin sync")
+		consensus.spinUpStateSync()
+	}
+
 	consensus.mutex.Lock()
 	defer consensus.mutex.Unlock()
 
@@ -236,10 +238,6 @@ func (consensus *Consensus) onCommitted(msg *msg_pb.Message) {
 	if !consensus.isRightBlockNumCheck(recvMsg) {
 		return
 	}
-	if recvMsg.BlockNum > consensus.blockNum {
-		consensus.getLogger().Info().Msg("[OnCommitted] low consensus block number. Spin up state sync")
-		consensus.spinUpStateSync()
-	}
 
 	aggSig, mask, err := consensus.ReadSignatureBitmapPayload(recvMsg.Payload, 0)
 	if err != nil {
@@ -272,6 +270,11 @@ func (consensus *Consensus) onCommitted(msg *msg_pb.Message) {
 
 	consensus.FBFTLog.AddMessage(recvMsg)
 
+	if recvMsg.BlockNum > consensus.blockNum {
+		consensus.getLogger().Info().Msg("[OnCommitted] low consensus block number. Spin up state sync")
+		consensus.spinUpStateSync()
+	}
+
 	consensus.mutex.Lock()
 	defer consensus.mutex.Unlock()
 

consensus/view_change.go

@@ -131,9 +131,8 @@ func (consensus *Consensus) getNextViewID() (uint64, time.Duration) {
 	if curTimestamp <= blockTimestamp {
 		return consensus.fallbackNextViewID()
 	}
-	totalNode := consensus.Decider.ParticipantsCount()
+	// diff only increases
-	// diff is at least 1, and it won't exceed the totalNode
+	diff := uint64((curTimestamp - blockTimestamp) / viewChangeTimeout)
-	diff := uint64(((curTimestamp - blockTimestamp) / viewChangeTimeout) % int64(totalNode))
 	nextViewID := diff + consensus.GetCurBlockViewID()
 
 	consensus.getLogger().Info().