Broadcast vote power. (#4748)

* Broadcast vote power.
* Moved `GetLastSigningPower` from private api to public.
* Fix for view change sign power.
* Rebased and fixed conflicts.
* Fix for broadcasting consensus vote power.

api/proto/message/harmonymessage.proto

@@ -29,6 +29,7 @@ enum MessageType {
   DRAND_INIT = 10 [deprecated=true];
   DRAND_COMMIT = 11 [deprecated=true];
   LOTTERY_REQUEST = 12 [deprecated=true]; // it should be either ENTER or GETPLAYERS but it will be removed later.
+  LAST_SIGN_POWER = 13;
 }
 
 // This is universal message for all communication protocols.
@@ -47,6 +48,7 @@ message Message {
     ViewChangeRequest viewchange = 7;
     // Refactor this later after demo.
     LotteryRequest lottery_request = 8 [deprecated=true];
+    LastSignPowerBroadcast last_sign_power = 9;
   }
 }
 
@@ -116,3 +118,11 @@ message ViewChangeRequest {
   bytes m3_bitmap= 12;
   bytes prepared_block = 13;
 }
+
+message LastSignPowerBroadcast {
+  int64 prepare = 1;
+  int64 commit = 2;
+  int64 change = 3;
+  bytes sender_pubkey = 4;
+  uint32 shard_id = 5;
+}

consensus/consensus.go

@@ -138,6 +138,10 @@ type Consensus struct {
 	// Both flags only for initialization state.
 	start           bool
 	isInitialLeader bool
+
+	// value receives from
+	lastKnownSignPower int64
+	lastKnowViewChange int64
 }
 
 // Blockchain returns the blockchain.
@@ -395,6 +399,18 @@ func (consensus *Consensus) InitConsensusWithValidators() (err error) {
 	return nil
 }
 
+func (consensus *Consensus) SetLastKnownSignPower(signPower, viewChange int64) {
+	atomic.StoreInt64(&consensus.lastKnownSignPower, signPower)
+	atomic.StoreInt64(&consensus.lastKnowViewChange, viewChange)
+}
+
+func (consensus *Consensus) GetLastKnownSignPower() int64 {
+	if consensus.IsViewChangingMode() {
+		return atomic.LoadInt64(&consensus.lastKnowViewChange)
+	}
+	return atomic.LoadInt64(&consensus.lastKnownSignPower)
+}
+
 type downloadAsync struct {
 }
 

consensus/consensus_service.go

@@ -7,6 +7,7 @@ import (
 
 	"github.com/ethereum/go-ethereum/common"
 	bls_core "github.com/harmony-one/bls/ffi/go/bls"
+	"github.com/harmony-one/harmony/api/proto"
 	msg_pb "github.com/harmony-one/harmony/api/proto/message"
 	consensus_engine "github.com/harmony-one/harmony/consensus/engine"
 	"github.com/harmony-one/harmony/consensus/quorum"
@@ -17,8 +18,10 @@ import (
 	bls_cosi "github.com/harmony-one/harmony/crypto/bls"
 	"github.com/harmony-one/harmony/crypto/hash"
 	"github.com/harmony-one/harmony/internal/chain"
+	nodeconfig "github.com/harmony-one/harmony/internal/configs/node"
 	"github.com/harmony-one/harmony/internal/utils"
 	"github.com/harmony-one/harmony/multibls"
+	"github.com/harmony-one/harmony/p2p"
 	"github.com/harmony-one/harmony/shard"
 	"github.com/harmony-one/harmony/shard/committee"
 	"github.com/harmony-one/harmony/webhooks"
@@ -153,6 +156,53 @@ func (consensus *Consensus) updateBitmaps() {
 
 }
 
+func (consensus *Consensus) sendLastSignPower() {
+	if consensus.isLeader() {
+		k, err := consensus.getLeaderPrivateKey(consensus.getLeaderPubKey().Object)
+		if err != nil {
+			consensus.getLogger().Err(err).Msg("Leader not found in the committee")
+			return
+		}
+		comm := getOrZero(consensus.decider.CurrentTotalPower(quorum.Commit))
+		prep := getOrZero(consensus.decider.CurrentTotalPower(quorum.Prepare))
+		view := consensus.decider.ComputeTotalPowerByMask(consensus.vc.GetViewIDBitmap(consensus.current.viewChangingID)).Int64()
+		for view > 100 {
+			view /= 10
+		}
+		msg := &msg_pb.Message{
+			ServiceType: msg_pb.ServiceType_CONSENSUS,
+			Type:        msg_pb.MessageType_LAST_SIGN_POWER,
+			Request: &msg_pb.Message_LastSignPower{
+				LastSignPower: &msg_pb.LastSignPowerBroadcast{
+					Prepare:      prep,
+					Commit:       comm,
+					Change:       view,
+					SenderPubkey: k.Pub.Bytes[:],
+					ShardId:      consensus.ShardID,
+				},
+			},
+		}
+		marshaledMessage, err := consensus.signAndMarshalConsensusMessage(msg, k.Pri)
+		if err != nil {
+			consensus.getLogger().Err(err).
+				Msg("[constructNewViewMessage] failed to sign and marshal the new view message")
+			return
+		}
+		if comm == 0 && prep == 0 && view == 0 {
+			return
+		}
+		msgToSend := proto.ConstructConsensusMessage(marshaledMessage)
+		if err := consensus.msgSender.SendWithoutRetry(
+			[]nodeconfig.GroupID{
+				nodeconfig.NewGroupIDByShardID(nodeconfig.ShardID(consensus.ShardID))},
+			p2p.ConstructMessage(msgToSend),
+		); err != nil {
+			consensus.getLogger().Err(err).
+				Msg("[LastSignPower] could not send out the ViewChange message")
+		}
+	}
+}
+
 // ResetState resets the state of the consensus
 func (consensus *Consensus) resetState() {
 	consensus.switchPhase("ResetState", FBFTAnnounce)

consensus/consensus_v2.go

@@ -20,6 +20,7 @@ import (
 	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/numeric"
 	"github.com/harmony-one/harmony/p2p"
 	"github.com/harmony-one/harmony/shard"
 	"github.com/harmony-one/vdf/src/vdf_go"
@@ -89,6 +90,8 @@ func (consensus *Consensus) HandleMessageUpdate(ctx context.Context, peer libp2p
 	case t == msg_pb.MessageType_NEWVIEW:
 		members := consensus.decider.Participants()
 		fbftMsg, err = ParseNewViewMessage(msg, members)
+	case t == msg_pb.MessageType_LAST_SIGN_POWER:
+		return nil
 	default:
 		fbftMsg, err = consensus.parseFBFTMessage(msg)
 	}
@@ -820,6 +823,7 @@ func (consensus *Consensus) setupForNewConsensus(blk *types.Block, committedMsg
 	}
 	consensus.fBFTLog.PruneCacheBeforeBlock(blk.NumberU64())
 	consensus.resetState()
+	consensus.sendLastSignPower()
 }
 
 func (consensus *Consensus) postCatchup(initBN uint64) {
@@ -968,3 +972,10 @@ func (consensus *Consensus) DeleteMessagesLessThan(number uint64) {
 	defer consensus.mutex.Unlock()
 	consensus.fBFTLog.deleteMessagesLessThan(number)
 }
+
+func getOrZero(n *numeric.Dec, err error) int64 {
+	if n == nil {
+		return 0
+	}
+	return (*n).Mul(numeric.NewDec(100)).TruncateInt64()
+}

consensus/fbft_log.go

@@ -36,6 +36,7 @@ type FBFTMessage struct {
 	M3AggSig           *bls_core.Sign
 	M3Bitmap           *bls_cosi.Mask
 	Verified           bool
+	LastVotePower      int64
 }
 
 func (m *FBFTMessage) Hash() []byte {

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

@@ -71,6 +71,14 @@ func (v *uniformVoteWeight) IsQuorumAchievedByMask(mask *bls_cosi.Mask) bool {
 	return true
 }
 
+func (v *uniformVoteWeight) ComputeTotalPowerByMask(mask *bls_cosi.Mask) numeric.Dec {
+	if mask == nil {
+		return numeric.ZeroDec()
+	}
+	currentTotalPower := utils.CountOneBits(mask.Bitmap)
+	return numeric.NewDec(currentTotalPower)
+}
+
 // QuorumThreshold ..
 func (v *uniformVoteWeight) QuorumThreshold() numeric.Dec {
 	return numeric.NewDec(v.TwoThirdsSignersCount())

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

@@ -163,6 +163,14 @@ func (v *stakedVoteWeight) IsQuorumAchievedByMask(mask *bls_cosi.Mask) bool {
 	return (*currentTotalPower).GT(threshold)
 }
 
+func (v *stakedVoteWeight) ComputeTotalPowerByMask(mask *bls_cosi.Mask) numeric.Dec {
+	if mask == nil {
+		return numeric.ZeroDec()
+	}
+	currentTotalPower := v.computeTotalPowerByMask(mask)
+	return *currentTotalPower
+}
+
 // ComputeTotalPowerByMask computes the total power indicated by bitmap mask
 func (v *stakedVoteWeight) computeTotalPowerByMask(mask *bls_cosi.Mask) *numeric.Dec {
 	currentTotal := numeric.ZeroDec()

consensus/quorum/quorum.go

@@ -118,6 +118,7 @@ type Decider interface {
 	) (*votepower.Ballot, error)
 	IsQuorumAchieved(Phase) bool
 	IsQuorumAchievedByMask(mask *bls_cosi.Mask) bool
+	ComputeTotalPowerByMask(mask *bls_cosi.Mask) numeric.Dec
 	QuorumThreshold() numeric.Dec
 	IsAllSigsCollected() bool
 	ResetPrepareAndCommitVotes()

consensus/quorum/thread_safe_decider.go

@@ -169,3 +169,9 @@ func (a threadSafeDeciderImpl) IsQuorumAchieved(p Phase) bool {
 	defer a.mu.Unlock()
 	return a.decider.IsQuorumAchieved(p)
 }
+
+func (a threadSafeDeciderImpl) ComputeTotalPowerByMask(mask *bls.Mask) numeric.Dec {
+	a.mu.Lock()
+	defer a.mu.Unlock()
+	return a.decider.ComputeTotalPowerByMask(mask)
+}

consensus/view_change.go

@@ -424,6 +424,7 @@ func (consensus *Consensus) onViewChange(recvMsg *FBFTMessage) {
 			Msg("[onViewChange] process View Change message error")
 		return
 	}
+	consensus.sendLastSignPower()
 
 	// received enough view change messages, change state to normal consensus
 	if consensus.decider.IsQuorumAchievedByMask(consensus.vc.GetViewIDBitmap(recvMsg.ViewID)) && consensus.isViewChangingMode() {

hmy/hmy.go

@@ -120,7 +120,6 @@ type NodeAPI interface {
 	GetConfig() commonRPC.Config
 	ShutDown()
 	GetLastSigningPower() (float64, error)
-	GetLastSigningPower2() (float64, error)
 }
 
 // New creates a new Harmony object (including the

node/harmony/api.go

@@ -1,10 +1,7 @@
 package node
 
 import (
-	"github.com/harmony-one/harmony/consensus/quorum"
-	"github.com/harmony-one/harmony/consensus/votepower"
 	"github.com/harmony-one/harmony/core/types"
-	"github.com/harmony-one/harmony/crypto/bls"
 	"github.com/harmony-one/harmony/eth/rpc"
 	"github.com/harmony-one/harmony/hmy"
 	"github.com/harmony-one/harmony/internal/tikv"
@@ -171,37 +168,7 @@ func (node *Node) GetConfig() rpc_common.Config {
 
 // GetLastSigningPower get last signed power
 func (node *Node) GetLastSigningPower() (float64, error) {
-	power, err := node.Consensus.Decider().CurrentTotalPower(quorum.Commit)
-	if err != nil {
-		return 0, err
-	}
-
-	round := float64(power.MulInt64(10000).RoundInt64()) / 10000
-	return round, nil
-}
-
-func (node *Node) GetLastSigningPower2() (float64, error) {
-	bc := node.Consensus.Blockchain()
-	cur := bc.CurrentBlock()
-	ss, err := bc.ReadShardState(cur.Epoch())
-	if err != nil {
-		return 0, err
-	}
-	roster, err := votepower.Compute(&ss.Shards[bc.ShardID()], cur.Epoch())
-	if err != nil {
-		return 0, err
-	}
-	blsPubKeys, err := ss.Shards[bc.ShardID()].BLSPublicKeys()
-	if err != nil {
-		return 0, err
-	}
-
-	mask := bls.NewMask(blsPubKeys)
-	err = mask.SetMask(cur.Header().LastCommitBitmap())
-	if err != nil {
-		return 0, err
-	}
-	power := roster.VotePowerByMask(mask)
-	round := float64(power.MulInt64(10000).RoundInt64()) / 10000
+	p := node.Consensus.GetLastKnownSignPower()
+	round := float64(p) / 100
 	return round, nil
 }

node/harmony/node.go

@@ -528,11 +528,16 @@ func validateShardBoundMessage(consensus *consensus.Consensus, peer libp2p_peer.
 		}
 	}
 
-	maybeCon, maybeVC := m.GetConsensus(), m.GetViewchange()
-	senderKey := []byte{}
-	senderBitmap := []byte{}
+	var (
+		maybeCon     = m.GetConsensus()
+		maybeVC      = m.GetViewchange()
+		maybeSP      = m.GetLastSignPower()
+		senderKey    []byte
+		senderBitmap []byte
+	)
 
-	if maybeCon != nil {
+	switch {
+	case maybeCon != nil:
 		if maybeCon.ShardId != consensus.ShardID {
 			nodeConsensusMessageCounterVec.With(prometheus.Labels{"type": "invalid_shard"}).Inc()
 			return nil, nil, true, errors.WithStack(errWrongShardID)
@@ -546,7 +551,7 @@ func validateShardBoundMessage(consensus *consensus.Consensus, peer libp2p_peer.
 		if maybeCon.ViewId+5 < consensus.GetCurBlockViewID() {
 			return nil, nil, true, errors.WithStack(errViewIDTooOld)
 		}
-	} else if maybeVC != nil {
+	case maybeVC != nil:
 		if maybeVC.ShardId != consensus.ShardID {
 			nodeConsensusMessageCounterVec.With(prometheus.Labels{"type": "invalid_shard"}).Inc()
 			return nil, nil, true, errors.WithStack(errWrongShardID)
@@ -556,7 +561,14 @@ func validateShardBoundMessage(consensus *consensus.Consensus, peer libp2p_peer.
 		if maybeVC.ViewId+5 < consensus.GetViewChangingID() {
 			return nil, nil, true, errors.WithStack(errViewIDTooOld)
 		}
-	} else {
+	case maybeSP != nil:
+		if maybeSP.ShardId != consensus.ShardID {
+			nodeConsensusMessageCounterVec.With(prometheus.Labels{"type": "invalid_shard"}).Inc()
+			return nil, nil, true, errors.WithStack(errWrongShardID)
+		}
+		senderKey = maybeSP.SenderPubkey
+		consensus.SetLastKnownSignPower(maybeSP.Commit, maybeSP.Change)
+	default:
 		nodeConsensusMessageCounterVec.With(prometheus.Labels{"type": "invalid"}).Inc()
 		return nil, nil, true, errors.WithStack(errNoSenderPubKey)
 	}

rpc/harmony/private_debug.go

@@ -58,17 +58,3 @@ func (s *PrivateDebugService) GetConfig(
 ) (StructuredResponse, error) {
 	return NewStructuredResponse(s.hmy.NodeAPI.GetConfig())
 }
-
-// GetLastSigningPower get last signed power
-func (s *PrivateDebugService) GetLastSigningPower(
-	ctx context.Context,
-) (float64, error) {
-	return s.hmy.NodeAPI.GetLastSigningPower()
-}
-
-// GetLastSigningPower2 get last signed power
-func (s *PrivateDebugService) GetLastSigningPower2(
-	ctx context.Context,
-) (float64, error) {
-	return s.hmy.NodeAPI.GetLastSigningPower2()
-}

rpc/harmony/public_debug.go

@@ -36,3 +36,10 @@ func (s *PublicDebugService) SetLogVerbosity(ctx context.Context, level int) (ma
 	utils.SetLogVerbosity(verbosity)
 	return map[string]interface{}{"verbosity": verbosity.String()}, nil
 }
+
+// GetLastSigningPower get last signed power
+func (s *PublicDebugService) GetLastSigningPower(
+	ctx context.Context,
+) (float64, error) {
+	return s.hmy.NodeAPI.GetLastSigningPower()
+}