merge vote accts with validator history

keepers/validator-keeper/src/lib.rs

@@ -8,6 +8,7 @@ use keeper_core::{get_vote_accounts_with_retry, CreateUpdateStats, SubmitStats};
 use log::error;
 use solana_account_decoder::UiDataSliceConfig;
 use solana_client::{
+    client_error::ClientError,
     nonblocking::rpc_client::RpcClient,
     rpc_config::{RpcAccountInfoConfig, RpcProgramAccountsConfig},
     rpc_filter::{Memcmp, RpcFilterType},
@@ -108,28 +109,7 @@ pub async fn emit_validator_history_metrics(
 ) -> Result<(), Box<dyn std::error::Error>> {
     let epoch = client.get_epoch_info().await?;
 
-    // Fetch every ValidatorHistory account
-    let gpa_config = RpcProgramAccountsConfig {
-        filters: Some(vec![RpcFilterType::Memcmp(Memcmp::new_raw_bytes(
-            0,
-            ValidatorHistory::discriminator().into(),
-        ))]),
-        account_config: RpcAccountInfoConfig {
-            encoding: Some(solana_account_decoder::UiAccountEncoding::Base64),
-            ..RpcAccountInfoConfig::default()
-        },
-        ..RpcProgramAccountsConfig::default()
-    };
-    let mut validator_history_accounts = client
-        .get_program_accounts_with_config(&program_id, gpa_config)
-        .await?;
-
-    let validator_histories = validator_history_accounts
-        .iter_mut()
-        .filter_map(|(_, account)| {
-            ValidatorHistory::try_deserialize(&mut account.data.as_slice()).ok()
-        })
-        .collect::<Vec<_>>();
+    let validator_histories = get_validator_history_accounts(client, program_id).await?;
 
     let mut ips = 0;
     let mut versions = 0;
@@ -213,6 +193,48 @@ pub async fn emit_validator_history_metrics(
     Ok(())
 }
 
+pub async fn get_validator_history_accounts(
+    client: &RpcClient,
+    program_id: Pubkey,
+) -> Result<Vec<ValidatorHistory>, ClientError> {
+    let gpa_config = RpcProgramAccountsConfig {
+        filters: Some(vec![RpcFilterType::Memcmp(Memcmp::new_raw_bytes(
+            0,
+            ValidatorHistory::discriminator().into(),
+        ))]),
+        account_config: RpcAccountInfoConfig {
+            encoding: Some(solana_account_decoder::UiAccountEncoding::Base64),
+            ..RpcAccountInfoConfig::default()
+        },
+        ..RpcProgramAccountsConfig::default()
+    };
+    let mut validator_history_accounts = client
+        .get_program_accounts_with_config(&program_id, gpa_config)
+        .await?;
+
+    let validator_histories = validator_history_accounts
+        .iter_mut()
+        .filter_map(|(_, account)| {
+            ValidatorHistory::try_deserialize(&mut account.data.as_slice()).ok()
+        })
+        .collect::<Vec<_>>();
+
+    Ok(validator_histories)
+}
+
+pub async fn get_validator_history_accounts_with_retry(
+    client: &RpcClient,
+    program_id: Pubkey,
+) -> Result<Vec<ValidatorHistory>, ClientError> {
+    for _ in 0..4 {
+        match get_validator_history_accounts(client, program_id).await {
+            Ok(validator_histories) => return Ok(validator_histories),
+            Err(_) => {}
+        }
+    }
+    get_validator_history_accounts(client, program_id).await
+}
+
 pub fn start_spy_server(
     cluster_entrypoint: SocketAddr,
     gossip_port: u16,

keepers/validator-keeper/src/vote_account.rs

@@ -1,3 +1,4 @@
+use std::collections::HashSet;
 use std::str::FromStr;
 use std::sync::Arc;
 
@@ -17,6 +18,8 @@ use validator_history::constants::{MAX_ALLOC_BYTES, MIN_VOTE_EPOCHS};
 use validator_history::state::ValidatorHistory;
 use validator_history::Config;
 
+use crate::get_validator_history_accounts_with_retry;
+
 #[derive(ThisError, Debug)]
 pub enum UpdateCommissionError {
     #[error(transparent)]
@@ -36,20 +39,14 @@ pub struct CopyVoteAccountEntry {
 }
 
 impl CopyVoteAccountEntry {
-    pub fn new(vote_account: &RpcVoteAccountInfo, program_id: &Pubkey, signer: &Pubkey) -> Self {
-        let vote_account = Pubkey::from_str(&vote_account.vote_pubkey)
-            .map_err(|e| {
-                error!("Invalid vote account pubkey");
-                e
-            })
-            .expect("Invalid vote account pubkey");
+    pub fn new(vote_account: &Pubkey, program_id: &Pubkey, signer: &Pubkey) -> Self {
         let (validator_history_account, _) = Pubkey::find_program_address(
             &[ValidatorHistory::SEED, &vote_account.to_bytes()],
             program_id,
         );
         let (config_address, _) = Pubkey::find_program_address(&[Config::SEED], program_id);
         Self {
-            vote_account,
+            vote_account: *vote_account,
             validator_history_account,
             config_address,
             program_id: *program_id,
@@ -119,10 +116,29 @@ pub async fn update_vote_accounts(
 ) -> Result<CreateUpdateStats, (UpdateCommissionError, CreateUpdateStats)> {
     let stats = CreateUpdateStats::default();
 
-    let vote_accounts = get_vote_accounts_with_retry(&rpc_client, MIN_VOTE_EPOCHS, None)
+    let rpc_vote_accounts = get_vote_accounts_with_retry(&rpc_client, MIN_VOTE_EPOCHS, None)
         .await
         .map_err(|e| (e.into(), stats))?;
 
+    let validator_histories =
+        get_validator_history_accounts_with_retry(&rpc_client, validator_history_program_id)
+            .await
+            .map_err(|e| (e.into(), stats))?;
+
+    // Merges new and active RPC vote accounts with all validator history accounts, and dedupes
+    let vote_accounts = rpc_vote_accounts
+        .iter()
+        .filter_map(|rpc_va| {
+            Pubkey::from_str(&rpc_va.vote_pubkey)
+                .map_err(|e| {
+                    error!("Invalid vote account pubkey");
+                    e
+                })
+                .ok()
+        })
+        .chain(validator_histories.iter().map(|vh| vh.vote_account))
+        .collect::<HashSet<_>>();
+
     let entries = vote_accounts
         .iter()
         .map(|va| CopyVoteAccountEntry::new(va, &validator_history_program_id, &keypair.pubkey()))