[stake][swarm] Add e2e rewards test (aptos-labs#3066)

### Description

Added rewards test:
- we have 4 nodes.
- one leaves validator set
- one is unreliable

We fetch their stakes, and confirm increases match their proposal success rate.


### Test Plan
added test

crates/aptos/src/node/analyze/analyze_validators.rs

@@ -14,6 +14,8 @@ use std::convert::TryFrom;
 use std::ops::Add;
 use storage_interface::{DbReader, Order};
 
+use super::fetch_metadata::ValidatorInfo;
+
 /// Single validator stats
 #[derive(Debug, Copy, Clone, PartialEq)]
 pub struct ValidatorStats {
@@ -25,6 +27,8 @@ pub struct ValidatorStats {
     pub votes: u32,
     /// Number of transactions in a block
     pub transactions: u32,
+    /// Voting power
+    pub voting_power: u64,
 }
 
 impl ValidatorStats {
@@ -94,6 +98,7 @@ impl Add for ValidatorStats {
             proposal_failures: self.proposal_failures + other.proposal_failures,
             votes: self.votes + other.votes,
             transactions: self.transactions + other.transactions,
+            voting_power: 0, // cannot aggregate voting power.
         }
     }
 }
@@ -152,6 +157,7 @@ impl Add for EpochStats {
                         proposal_successes: 0,
                         votes: 0,
                         transactions: 0,
+                        voting_power: 0,
                     }),
             );
         }
@@ -221,25 +227,33 @@ impl AnalyzeValidators {
     }
 
     /// Analyze single epoch
-    pub fn analyze(
-        mut blocks: Vec<VersionedNewBlockEvent>,
-        validators: &[AccountAddress],
-    ) -> EpochStats {
+    pub fn analyze(blocks: &[VersionedNewBlockEvent], validators: &[ValidatorInfo]) -> EpochStats {
         assert!(
             validators.iter().as_slice().windows(2).all(|w| {
-                w[0].partial_cmp(&w[1])
+                w[0].validator_index
+                    .partial_cmp(&w[1].validator_index)
                     .map(|o| o != Ordering::Greater)
                     .unwrap_or(false)
             }),
             "Validators need to be sorted"
         );
-        blocks.sort_by_key(|e| e.event.round());
+        assert!(
+            blocks.iter().as_slice().windows(2).all(|w| {
+                w[0].event
+                    .round()
+                    .partial_cmp(&w[1].event.round())
+                    .map(|o| o != Ordering::Greater)
+                    .unwrap_or(false)
+            }),
+            "Blocks need to be sorted"
+        );
 
         let mut successes = HashMap::<AccountAddress, u32>::new();
         let mut failures = HashMap::<AccountAddress, u32>::new();
         let mut votes = HashMap::<AccountAddress, u32>::new();
         let mut transactions = HashMap::<AccountAddress, u32>::new();
 
+        let mut trimmed_rounds = 0;
         let mut nil_blocks = 0;
         let mut previous_round = 0;
         for (pos, block) in blocks.iter().enumerate() {
@@ -248,13 +262,16 @@ impl AnalyzeValidators {
             if is_nil {
                 nil_blocks += 1;
             }
-            if event.round() + (if is_nil { 1 } else { 0 })
-                != previous_round + 1 + event.failed_proposer_indices().len() as u64
-            {
+            let expected_round = previous_round
+                + (if is_nil { 0 } else { 1 })
+                + event.failed_proposer_indices().len() as u64;
+            if event.round() != expected_round {
                 println!(
                     "Missing failed AccountAddresss : {} {:?}",
                     previous_round, &event
                 );
+                assert!(expected_round < event.round());
+                trimmed_rounds += event.round() - expected_round;
             }
             previous_round = event.round();
 
@@ -264,7 +281,7 @@ impl AnalyzeValidators {
 
             for failed_proposer_index in event.failed_proposer_indices() {
                 *failures
-                    .entry(validators[*failed_proposer_index as usize])
+                    .entry(validators[*failed_proposer_index as usize].address)
                     .or_insert(0) += 1;
             }
 
@@ -276,7 +293,7 @@ impl AnalyzeValidators {
             );
             for (i, validator) in validators.iter().enumerate() {
                 if previous_block_votes_bitvec.is_set(i as u16) {
-                    *votes.entry(*validator).or_insert(0) += 1;
+                    *votes.entry(validator.address).or_insert(0) += 1;
                 }
             }
 
@@ -301,22 +318,27 @@ impl AnalyzeValidators {
         let total_transactions: u32 = transactions.values().sum();
         let total_rounds = total_successes + total_failures;
         assert_eq!(
-            total_rounds, previous_round as u32,
-            "{} {}",
-            total_rounds, previous_round
+            total_rounds + u32::try_from(trimmed_rounds).unwrap(),
+            previous_round as u32,
+            "{} success + {} failures + {} trimmed != {}",
+            total_successes,
+            total_failures,
+            trimmed_rounds,
+            previous_round
         );
 
         return EpochStats {
             validator_stats: validators
                 .iter()
                 .map(|validator| {
                     (
-                        *validator,
+                        validator.address,
                         ValidatorStats {
-                            proposal_successes: *successes.get(validator).unwrap_or(&0),
-                            proposal_failures: *failures.get(validator).unwrap_or(&0),
-                            votes: *votes.get(validator).unwrap_or(&0),
-                            transactions: *transactions.get(validator).unwrap_or(&0),
+                            proposal_successes: *successes.get(&validator.address).unwrap_or(&0),
+                            proposal_failures: *failures.get(&validator.address).unwrap_or(&0),
+                            votes: *votes.get(&validator.address).unwrap_or(&0),
+                            transactions: *transactions.get(&validator.address).unwrap_or(&0),
+                            voting_power: validator.voting_power,
                         },
                     )
                 })
@@ -332,7 +354,7 @@ impl AnalyzeValidators {
     /// Print validator stats in a table
     pub fn print_detailed_epoch_table(
         epoch_stats: &EpochStats,
-        extra: Option<&HashMap<AccountAddress, &str>>,
+        extra: Option<(&str, &HashMap<AccountAddress, String>)>,
         sort_by_health: bool,
     ) {
         println!(
@@ -342,8 +364,15 @@ impl AnalyzeValidators {
             100.0 * epoch_stats.nil_blocks as f32 / epoch_stats.total_rounds as f32,
         );
         println!(
-            "{: <10} | {: <10} | {: <10} | {: <10} | {: <10} | {: <10} | {: <10} | ",
-            "elected", "% rounds", "% failed", "succeded", "failed", "voted", "transact"
+            "{: <10} | {: <10} | {: <10} | {: <10} | {: <10} | {: <10} | {: <10} | {: <30}",
+            "elected",
+            "% rounds",
+            "% failed",
+            "succeded",
+            "failed",
+            "voted",
+            "transact",
+            extra.map(|(column, _)| column).unwrap_or("")
         );
 
         let mut validator_order: Vec<&AccountAddress> =
@@ -382,10 +411,10 @@ impl AnalyzeValidators {
                 cur_stats.proposal_failures,
                 cur_stats.votes,
                 cur_stats.transactions,
-                if let Some(extra_map) = extra {
+                if let Some((_, extra_map)) = extra {
                     format!(
                         "{: <30} | {}",
-                        extra_map.get(validator).unwrap_or(&""),
+                        extra_map.get(validator).unwrap_or(&"".to_string()),
                         validator
                     )
                 } else {

crates/aptos/src/node/analyze/fetch_metadata.rs

@@ -10,10 +10,17 @@ use aptos_types::account_address::AccountAddress;
 use std::convert::TryFrom;
 use std::str::FromStr;
 
+#[derive(Eq, PartialEq, Clone, Copy, Debug)]
+pub struct ValidatorInfo {
+    pub address: AccountAddress,
+    pub voting_power: u64,
+    pub validator_index: u64,
+}
+
 pub struct EpochInfo {
     pub epoch: u64,
     pub blocks: Vec<VersionedNewBlockEvent>,
-    pub validators: Vec<AccountAddress>,
+    pub validators: Vec<ValidatorInfo>,
 }
 
 pub struct FetchMetadata {}
@@ -22,18 +29,30 @@ impl FetchMetadata {
     fn get_validator_addresses(
         data: &MoveResource,
         field_name: &str,
-    ) -> Result<Vec<AccountAddress>> {
-        fn extract_validator_address(validator: &serde_json::Value) -> Result<AccountAddress> {
-            if let serde_json::Value::Object(value) = validator {
-                if let Some(serde_json::Value::String(address)) = &value.get("addr") {
-                    AccountAddress::from_hex_literal(address)
-                        .map_err(|e| anyhow!("Cannot parse address {:?}", e))
-                } else {
-                    Err(anyhow!("Addr not present or of correct type"))
-                }
-            } else {
-                Err(anyhow!("Validator config not a json object"))
-            }
+    ) -> Result<Vec<ValidatorInfo>> {
+        fn extract_validator_address(validator: &serde_json::Value) -> Result<ValidatorInfo> {
+            Ok(ValidatorInfo {
+                address: AccountAddress::from_hex_literal(
+                    validator.get("addr").unwrap().as_str().unwrap(),
+                )
+                .map_err(|e| anyhow!("Cannot parse address {:?}", e))?,
+                voting_power: validator
+                    .get("voting_power")
+                    .unwrap()
+                    .as_str()
+                    .unwrap()
+                    .parse()
+                    .map_err(|e| anyhow!("Cannot parse voting_power {:?}", e))?,
+                validator_index: validator
+                    .get("config")
+                    .unwrap()
+                    .get("validator_index")
+                    .unwrap()
+                    .as_str()
+                    .unwrap()
+                    .parse()
+                    .map_err(|e| anyhow!("Cannot parse validator_index {:?}", e))?,
+            })
         }
 
         let validators_json = data
@@ -42,7 +61,7 @@ impl FetchMetadata {
             .get(&IdentifierWrapper::from_str(field_name).unwrap())
             .unwrap();
         if let serde_json::Value::Array(validators_array) = validators_json {
-            let mut validators: Vec<AccountAddress> = vec![];
+            let mut validators: Vec<ValidatorInfo> = vec![];
             for validator in validators_array {
                 validators.push(extract_validator_address(validator)?);
             }
@@ -52,21 +71,21 @@ impl FetchMetadata {
         }
     }
 
-    fn get_validators_from_transaction(transaction: &Transaction) -> Result<Vec<AccountAddress>> {
+    fn get_validators_from_transaction(transaction: &Transaction) -> Result<Vec<ValidatorInfo>> {
         if let Ok(info) = transaction.transaction_info() {
             for change in &info.changes {
                 if let WriteSetChange::WriteResource(resource) = change {
                     if resource.data.typ.name.0.as_str() == "ValidatorSet" {
                         // No pending at epoch change
                         assert_eq!(
-                            Vec::<AccountAddress>::new(),
+                            Vec::<ValidatorInfo>::new(),
                             FetchMetadata::get_validator_addresses(
                                 &resource.data,
                                 "pending_inactive"
                             )?
                         );
                         assert_eq!(
-                            Vec::<AccountAddress>::new(),
+                            Vec::<ValidatorInfo>::new(),
                             FetchMetadata::get_validator_addresses(
                                 &resource.data,
                                 "pending_active"
@@ -89,13 +108,12 @@ impl FetchMetadata {
         end_epoch: Option<u64>,
     ) -> Result<Vec<EpochInfo>> {
         let mut start_seq_num = 0;
-        let last_seq_num = client
+        let (last_events, state) = client
             .get_new_block_events(None, Some(1))
             .await?
-            .into_inner()
-            .first()
-            .unwrap()
-            .sequence_number;
+            .into_parts();
+        assert_eq!(last_events.len(), 1, "{:?}", last_events);
+        let last_seq_num = last_events.first().unwrap().sequence_number;
 
         if let Some(start_epoch) = start_epoch {
             if start_epoch > 1 {
@@ -130,11 +148,11 @@ impl FetchMetadata {
         let mut batch_index = 0;
 
         println!(
-            "Fetching {} to {} sequence number",
-            start_seq_num, last_seq_num
+            "Fetching {} to {} sequence number, last version: {} and epoch: {}",
+            start_seq_num, last_seq_num, state.version, state.epoch,
         );
 
-        let mut validators: Vec<AccountAddress> = vec![];
+        let mut validators: Vec<ValidatorInfo> = vec![];
         let mut epoch = 0;
 
         let mut current: Vec<VersionedNewBlockEvent> = vec![];
@@ -194,7 +212,7 @@ impl FetchMetadata {
                                     current = vec![];
 
                                     validators = new_validators;
-                                    validators.sort();
+                                    validators.sort_by_key(|v| v.validator_index);
                                     assert_eq!(epoch + 1, event.event.epoch());
                                     epoch = event.event.epoch();
                                     if end_epoch.is_some() && epoch >= end_epoch.unwrap() {

crates/aptos/src/node/mod.rs

@@ -767,10 +767,22 @@ impl CliCommand<()> for AnalyzeValidatorPerformance {
         let print_detailed = self.analyze_mode == AnalyzeMode::DetailedEpochTable
             || self.analyze_mode == AnalyzeMode::All;
         for epoch_info in epochs {
-            let epoch_stats = AnalyzeValidators::analyze(epoch_info.blocks, &epoch_info.validators);
+            let epoch_stats =
+                AnalyzeValidators::analyze(&epoch_info.blocks, &epoch_info.validators);
             if print_detailed {
                 println!("Detailed table for epoch {}:", epoch_info.epoch);
-                AnalyzeValidators::print_detailed_epoch_table(&epoch_stats, None, true);
+                AnalyzeValidators::print_detailed_epoch_table(
+                    &epoch_stats,
+                    Some((
+                        "voting_power",
+                        &epoch_info
+                            .validators
+                            .iter()
+                            .map(|v| (v.address, v.voting_power.to_string()))
+                            .collect::<HashMap<_, _>>(),
+                    )),
+                    true,
+                );
             }
             stats.insert(epoch_info.epoch, epoch_stats);
         }

crates/aptos/src/test/mod.rs

@@ -44,6 +44,7 @@ use aptos_types::{
     validator_info::ValidatorInfo,
 };
 use reqwest::Url;
+use serde::{Deserialize, Serialize};
 use serde_json::Value;
 use std::collections::HashMap;
 use std::{collections::BTreeMap, path::PathBuf, str::FromStr, time::Duration};
@@ -858,7 +859,7 @@ pub struct ValidatorSet {
     pub pending_active: Vec<ValidatorInfo>,
 }
 
-fn to_validator_set(value: &serde_json::Value) -> ValidatorSet {
+pub fn to_validator_set(value: &serde_json::Value) -> ValidatorSet {
     ValidatorSet {
         consensus_scheme: match value.get("consensus_scheme").unwrap().as_u64().unwrap() {
             0u64 => ConsensusScheme::BLS12381,
@@ -886,3 +887,24 @@ fn json_account_to_balance(value: &Value) -> u64 {
     )
     .unwrap()
 }
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct IndividualValidatorPerformance {
+    successful_proposals: String,
+    failed_proposals: String,
+}
+
+impl IndividualValidatorPerformance {
+    pub fn successful_proposals(&self) -> u32 {
+        self.successful_proposals.parse().unwrap()
+    }
+
+    pub fn failed_proposals(&self) -> u32 {
+        self.failed_proposals.parse().unwrap()
+    }
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct ValidatorPerformance {
+    pub validators: Vec<IndividualValidatorPerformance>,
+}