Leader rotation. Check next leader aliveness. (#4359)

* Cleanup and fix update pub keys.

* Skip the next leader if it doesn't sign blocks.

* Comment & constant.

* Updated with dev.

* Updated with latest dev.

* Cleanup

cmd/harmony/main.go

@@ -788,6 +788,8 @@ func setupChain(hc harmonyconfig.HarmonyConfig, nodeConfig *nodeconfig.ConfigTyp
 }
 
 func setupConsensusAndNode(hc harmonyconfig.HarmonyConfig, nodeConfig *nodeconfig.ConfigType, registry *registry.Registry) *node.Node {
+	decider := quorum.NewDecider(quorum.SuperMajorityVote, uint32(hc.General.ShardID))
+
 	// Parse minPeers from harmonyconfig.HarmonyConfig
 	var minPeers int
 	var aggregateSig bool
@@ -821,7 +823,6 @@ func setupConsensusAndNode(hc harmonyconfig.HarmonyConfig, nodeConfig *nodeconfi
 	registry.SetCxPool(cxPool)
 
 	// Consensus object.
-	decider := quorum.NewDecider(quorum.SuperMajorityVote, nodeConfig.ShardID)
 	registry.SetIsBackup(isBackup(hc))
 	currentConsensus, err := consensus.New(
 		myHost, nodeConfig.ShardID, nodeConfig.ConsensusPriKey, registry, decider, minPeers, aggregateSig)

consensus/consensus_v2.go

@@ -690,10 +690,15 @@ func (consensus *Consensus) commitBlock(blk *types.Block, committedMsg *FBFTMess
 // This function must be called with enabled leader rotation.
 func (consensus *Consensus) rotateLeader(epoch *big.Int) {
 	var (
-		bc     = consensus.Blockchain()
-		prev   = consensus.getLeaderPubKey()
-		leader = consensus.getLeaderPubKey()
+		bc        = consensus.Blockchain()
+		prev      = consensus.getLeaderPubKey()
+		leader    = consensus.getLeaderPubKey()
+		curBlock  = bc.CurrentBlock()
+		curNumber = curBlock.NumberU64()
+		curEpoch  = curBlock.Epoch().Uint64()
 	)
+	const blocksCountAliveness = 10
+
 	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 {
@@ -741,18 +746,59 @@ func (consensus *Consensus) rotateLeader(epoch *big.Int) {
 	var (
 		wasFound bool
 		next     *bls.PublicKeyWrapper
+		offset   = 1
 	)
-	if bc.Config().IsLeaderRotationExternalValidatorsAllowed(epoch) {
-		wasFound, next = consensus.Decider.NthNextValidator(committee.Slots, leader, 1)
-	} else {
-		wasFound, next = consensus.Decider.NthNextHmy(shard.Schedule.InstanceForEpoch(epoch), leader, 1)
-	}
-	if !wasFound {
-		utils.Logger().Error().Msg("Failed to get next leader")
-		return
-	} else {
+
+	for {
+		if bc.Config().IsLeaderRotationExternalValidatorsAllowed(epoch) {
+			wasFound, next = consensus.Decider.NthNextValidator(committee.Slots, leader, offset)
+		} else {
+			wasFound, next = consensus.Decider.NthNextHmy(shard.Schedule.InstanceForEpoch(epoch), leader, offset)
+		}
+		if !wasFound {
+			utils.Logger().Error().Msg("Failed to get next leader")
+			// Seems like nothing we can do here.
+			return
+		}
+		members := consensus.Decider.Participants()
+		mask := bls.NewMask(members)
+		skipped := 0
+		for i := 0; i < blocksCountAliveness; i++ {
+			header := bc.GetHeaderByNumber(curNumber - uint64(i))
+			if header == nil {
+				utils.Logger().Error().Msgf("Failed to get header by number %d", curNumber-uint64(i))
+				return
+			}
+			// if epoch is different, we should not check this block.
+			if header.Epoch().Uint64() != curEpoch {
+				break
+			}
+			// Populate the mask with the bitmap.
+			err = mask.SetMask(header.LastCommitBitmap())
+			if err != nil {
+				utils.Logger().Err(err).Msg("Failed to set mask")
+				return
+			}
+			ok, err := mask.KeyEnabled(next.Bytes)
+			if err != nil {
+				utils.Logger().Err(err).Msg("Failed to get key enabled")
+				return
+			}
+			if !ok {
+				skipped++
+			}
+		}
+
+		// no signature from the next leader at all, we should skip it.
+		if skipped >= blocksCountAliveness {
+			// Next leader is not signing blocks, we should skip it.
+			offset++
+			continue
+		}
 		consensus.setLeaderPubKey(next)
+		break
 	}
+
 	if consensus.isLeader() && !consensus.getLeaderPubKey().Object.IsEqual(prev.Object) {
 		// leader changed
 		go func() {

consensus/quorum/quorum.go

@@ -77,7 +77,7 @@ type ParticipantTracker interface {
 	ParticipantsCount() int64
 	// NthNextValidator returns key for next validator. It assumes external validators and leader rotation.
 	NthNextValidator(slotList shard.SlotList, pubKey *bls.PublicKeyWrapper, next int) (bool, *bls.PublicKeyWrapper)
-	NthNextHmy(shardingconfig.Instance, *bls.PublicKeyWrapper, int) (bool, *bls.PublicKeyWrapper)
+	NthNextHmy(instance shardingconfig.Instance, pubkey *bls.PublicKeyWrapper, next int) (bool, *bls.PublicKeyWrapper)
 	NthNextHmyExt(shardingconfig.Instance, *bls.PublicKeyWrapper, int) (bool, *bls.PublicKeyWrapper)
 	FirstParticipant(shardingconfig.Instance) *bls.PublicKeyWrapper
 	UpdateParticipants(pubKeys, allowlist []bls.PublicKeyWrapper)

internal/utils/math.go

@@ -0,0 +1,17 @@
+package utils
+
+import "golang.org/x/exp/constraints"
+
+func Min[T constraints.Ordered](a, b T) T {
+	if a < b {
+		return a
+	}
+	return b
+}
+
+func Max[T constraints.Ordered](a, b T) T {
+	if a > b {
+		return a
+	}
+	return b
+}