[reward][rpc] Extend ValidatorStats with earning per key, implement tracking rewards in payout, expose network wide (#2656)

* [reward] Implement tracking rewards for beacon & shardchain (via crosslinks)

* [reward][rpc] Track reward given out per key for validator stats, compose to expose in validator RPC

consensus/reward/rewarder.go

@@ -12,6 +12,7 @@ type Payout struct {
 	ShardID     uint32
 	Addr        common.Address
 	NewlyEarned *big.Int
+	EarningKey  shard.BLSPublicKey
 }
 
 // CompletedRound ..

core/blockchain.go

@@ -2313,7 +2313,14 @@ func (bc *BlockChain) UpdateValidatorVotingPower(
 			total = total.Add(value[i].EffectiveStake)
 		}
 		stats.TotalEffectiveStake = total
-		stats.MetricsPerShard = value
+		earningWrapping := make([]staking.VoteWithCurrentEpochEarning, len(value))
+		for i := range value {
+			earningWrapping[i] = staking.VoteWithCurrentEpochEarning{
+				VoteOnSubcomittee: value[i],
+				Earned:            common.Big0,
+			}
+		}
+		stats.MetricsPerShard = earningWrapping
 		wrapper, err := state.ValidatorWrapper(key)
 		if err != nil {
 			return err

core/offchain.go

@@ -208,11 +208,41 @@ func (bc *BlockChain) CommitOffChainData(
 	// Update block reward accumulator and slashes
 	if isBeaconChain {
 		if isStaking {
+			roundResult := payout.ReadRoundResult()
 			if err := bc.UpdateBlockRewardAccumulator(
-				batch, payout.ReadRoundResult().Total, block.Number().Uint64(),
+				batch, roundResult.Total, block.Number().Uint64(),
 			); err != nil {
 				return NonStatTy, err
 			}
+
+			for _, paid := range [...][]reward.Payout{
+				roundResult.BeaconchainAward, roundResult.ShardChainAward,
+			} {
+				for i := range paid {
+					if stats, err := bc.ReadValidatorStats(paid[i].Addr); err == nil {
+						doUpdate := false
+						for j := range stats.MetricsPerShard {
+							if stats.MetricsPerShard[j].Identity == paid[i].EarningKey {
+								doUpdate = true
+								stats.MetricsPerShard[j].Earned.Add(
+									stats.MetricsPerShard[j].Earned,
+									paid[i].NewlyEarned,
+								)
+							}
+						}
+						if !doUpdate {
+							if err := rawdb.WriteValidatorStats(batch, paid[i].Addr, stats); err != nil {
+								utils.Logger().Info().
+									Err(err).Msg("could not update earning per key in stats")
+							}
+						}
+					} else {
+						utils.Logger().Info().
+							Err(err).Msg("could not read validator stats to update for earning per key")
+					}
+				}
+			}
+
 			records := slash.Records{}
 			if s := header.Slashes(); len(s) > 0 {
 				if err := rlp.DecodeBytes(s, &records); err != nil {

internal/chain/reward.go

@@ -26,9 +26,6 @@ import (
 func ballotResultBeaconchain(
 	bc engine.ChainReader, header *block.Header,
 ) (shard.SlotList, shard.SlotList, shard.SlotList, error) {
-	// TODO ek – retrieving by parent number (blockNum - 1) doesn't work,
-	//  while it is okay with hash.  Sounds like DB inconsistency.
-	//  Figure out why.
 	parentHeader := bc.GetHeaderByHash(header.ParentHash())
 	if parentHeader == nil {
 		return nil, nil, nil, ctxerror.New(
@@ -98,7 +95,7 @@ func AccumulateRewards(
 	}
 
 	// After staking
-	if bc.Config().IsStaking(header.Epoch()) &&
+	if headerE := header.Epoch(); bc.Config().IsStaking(headerE) &&
 		bc.CurrentHeader().ShardID() == shard.BeaconChainShardID {
 		utils.AnalysisStart("accumulateRewardBeaconchainSelfPayout", nowEpoch, blockNow)
 		defaultReward := network.BaseStakedReward
@@ -127,7 +124,8 @@ func AccumulateRewards(
 			return network.EmptyPayout, nil
 		}
 
-		newRewards := big.NewInt(0)
+		newRewards, beaconP, shardP :=
+			big.NewInt(0), []reward.Payout{}, []reward.Payout{}
 
 		// Take care of my own beacon chain committee, _ is missing, for slashing
 		members, payable, missing, err := ballotResultBeaconchain(beaconChain, header)
@@ -147,17 +145,20 @@ func AccumulateRewards(
 		}
 		beaconCurrentEpoch := beaconChain.CurrentHeader().Epoch()
 		votingPower, err := lookupVotingPower(
-			header.Epoch(), beaconCurrentEpoch, &subComm,
+			headerE, beaconCurrentEpoch, &subComm,
 		)
 		if err != nil {
 			return network.EmptyPayout, err
 		}
 
-		beaconExternalShare := shard.Schedule.InstanceForEpoch(header.Epoch()).ExternalVotePercent()
+		beaconExternalShare := shard.Schedule.InstanceForEpoch(
+			headerE,
+		).ExternalVotePercent()
 		for beaconMember := range payable {
 			// TODO Give out whatever leftover to the last voter/handle
 			// what to do about share of those that didn't sign
-			voter := votingPower.Voters[payable[beaconMember].BLSPublicKey]
+			blsKey := payable[beaconMember].BLSPublicKey
+			voter := votingPower.Voters[blsKey]
 			if !voter.IsHarmonyNode {
 				snapshot, err := bc.ReadValidatorSnapshot(voter.EarningAccount)
 				if err != nil {
@@ -170,6 +171,12 @@ func AccumulateRewards(
 				if err := state.AddReward(snapshot, due); err != nil {
 					return network.EmptyPayout, err
 				}
+				beaconP = append(beaconP, reward.Payout{
+					ShardID:     shard.BeaconChainShardID,
+					Addr:        voter.EarningAccount,
+					NewlyEarned: due,
+					EarningKey:  voter.Identity,
+				})
 			}
 		}
 		utils.AnalysisEnd("accumulateRewardBeaconchainSelfPayout", nowEpoch, blockNow)
@@ -184,9 +191,10 @@ func AccumulateRewards(
 
 			type slotPayable struct {
 				shard.Slot
-				payout *big.Int
-				bucket int
-				index  int
+				payout  *big.Int
+				bucket  int
+				index   int
+				shardID uint32
 			}
 
 			type slotMissing struct {
@@ -237,7 +245,9 @@ func AccumulateRewards(
 					return network.EmptyPayout, err
 				}
 
-				shardExternalShare := shard.Schedule.InstanceForEpoch(cxLink.Epoch()).ExternalVotePercent()
+				shardExternalShare := shard.Schedule.InstanceForEpoch(
+					epoch,
+				).ExternalVotePercent()
 				for j := range payableSigners {
 					voter := votingPower.Voters[payableSigners[j].BLSPublicKey]
 					if !voter.IsHarmonyNode && !voter.OverallPercent.IsZero() {
@@ -245,10 +255,11 @@ func AccumulateRewards(
 							voter.OverallPercent.Quo(shardExternalShare),
 						)
 						allPayables = append(allPayables, slotPayable{
-							Slot:   payableSigners[j],
-							payout: due.TruncateInt(),
-							bucket: i,
-							index:  j,
+							Slot:    payableSigners[j],
+							payout:  due.TruncateInt(),
+							bucket:  i,
+							index:   j,
+							shardID: shardID,
 						})
 					}
 				}
@@ -284,8 +295,9 @@ func AccumulateRewards(
 			// Finally do the pay
 			for bucket := range resultsHandle {
 				for payThem := range resultsHandle[bucket] {
+					payable := resultsHandle[bucket][payThem]
 					snapshot, err := bc.ReadValidatorSnapshot(
-						resultsHandle[bucket][payThem].EcdsaAddress,
+						payable.EcdsaAddress,
 					)
 					if err != nil {
 						return network.EmptyPayout, err
@@ -295,10 +307,18 @@ func AccumulateRewards(
 					if err := state.AddReward(snapshot, due); err != nil {
 						return network.EmptyPayout, err
 					}
+					shardP = append(shardP, reward.Payout{
+						ShardID:     payable.shardID,
+						Addr:        payable.EcdsaAddress,
+						NewlyEarned: due,
+						EarningKey:  payable.BLSPublicKey,
+					})
 				}
 			}
 			utils.AnalysisEnd("accumulateRewardShardchainPayout", nowEpoch, blockNow)
-			return network.NewStakingEraRewardForRound(newRewards, missing), nil
+			return network.NewStakingEraRewardForRound(
+				newRewards, missing, beaconP, shardP,
+			), nil
 		}
 		return network.EmptyPayout, nil
 	}

staking/network/reward.go

@@ -79,18 +79,19 @@ type stakingEra struct {
 func NewStakingEraRewardForRound(
 	totalPayout *big.Int,
 	mia shard.SlotList,
+	beaconP, shardP []reward.Payout,
 ) reward.Reader {
-
 	return &stakingEra{
 		CompletedRound: reward.CompletedRound{
 			Total:            totalPayout,
-			BeaconchainAward: []reward.Payout{},
-			ShardChainAward:  []reward.Payout{},
+			BeaconchainAward: beaconP,
+			ShardChainAward:  shardP,
 		},
 		missingSigners: mia,
 	}
 }
 
+// MissingSigners ..
 func (r *stakingEra) MissingSigners() shard.SlotList {
 	return r.missingSigners
 }

staking/types/validator.go

@@ -121,7 +121,7 @@ func NewEmptyStats() *ValidatorStats {
 	return &ValidatorStats{
 		numeric.ZeroDec(),
 		numeric.ZeroDec(),
-		[]votepower.VoteOnSubcomittee{},
+		[]VoteWithCurrentEpochEarning{},
 	}
 }
 
@@ -167,6 +167,12 @@ func (w ValidatorWrapper) MarshalJSON() ([]byte, error) {
 	})
 }
 
+// VoteWithCurrentEpochEarning ..
+type VoteWithCurrentEpochEarning struct {
+	votepower.VoteOnSubcomittee
+	Earned *big.Int `json:"earned-reward"`
+}
+
 // ValidatorStats to record validator's performance and history records
 type ValidatorStats struct {
 	// APR ..
@@ -174,7 +180,7 @@ type ValidatorStats struct {
 	// TotalEffectiveStake is the total effective stake this validator has
 	TotalEffectiveStake numeric.Dec `json:"total-effective-stake"`
 	// MetricsPerShard ..
-	MetricsPerShard []votepower.VoteOnSubcomittee `json:"by-shard"`
+	MetricsPerShard []VoteWithCurrentEpochEarning `json:"by-shard"`
 }
 
 func (s ValidatorStats) String() string {