review comments

keepers/validator-keeper/src/lib.rs

@@ -80,6 +80,16 @@ pub fn emit_mev_commission_datapoint(stats: CreateUpdateStats) {
     );
 }
 
+pub fn emit_mev_earned_datapoint(stats: CreateUpdateStats) {
+    datapoint_info!(
+        "mev-earned-stats",
+        ("num_creates_success", stats.creates.successes, i64),
+        ("num_creates_error", stats.creates.errors, i64),
+        ("num_updates_success", stats.updates.successes, i64),
+        ("num_updates_error", stats.updates.errors, i64),
+    );
+}
+
 pub fn emit_validator_commission_datapoint(stats: CreateUpdateStats, runs_for_epoch: i64) {
     datapoint_info!(
         "vote-account-stats",

keepers/validator-keeper/src/main.rs

@@ -18,10 +18,10 @@ use solana_sdk::{
 use tokio::time::sleep;
 use validator_keeper::{
     cluster_info::update_cluster_info,
-    emit_cluster_history_datapoint, emit_mev_commission_datapoint,
+    emit_cluster_history_datapoint, emit_mev_commission_datapoint, emit_mev_earned_datapoint,
     emit_validator_commission_datapoint, emit_validator_history_metrics,
     gossip::{emit_gossip_datapoint, upload_gossip_values},
-    mev_commission::update_mev_commission,
+    mev_commission::{update_mev_commission, update_mev_earned},
     stake::{emit_stake_history_datapoint, update_stake_history},
     vote_account::update_vote_accounts,
 };
@@ -115,6 +115,43 @@ async fn mev_commission_loop(
     }
 }
 
+async fn mev_earned_loop(
+    client: Arc<RpcClient>,
+    keypair: Arc<Keypair>,
+    commission_history_program_id: Pubkey,
+    tip_distribution_program_id: Pubkey,
+    interval: u64,
+) {
+    let mut curr_epoch = 0;
+    // {TipDistributionAccount : VoteAccount}
+    let mut validators_updated: HashMap<Pubkey, Pubkey> = HashMap::new();
+
+    loop {
+        // Continuously runs throughout an epoch, polling for tip distribution accounts from the prev epoch with uploaded merkle roots
+        // and submitting update_mev_earned (technically update_mev_comission) txs when the uploaded merkle roots are detected
+        match update_mev_earned(
+            client.clone(),
+            keypair.clone(),
+            &commission_history_program_id,
+            &tip_distribution_program_id,
+            &mut validators_updated,
+            &mut curr_epoch,
+        )
+        .await
+        {
+            Ok(stats) => {
+                emit_mev_earned_datapoint(stats);
+                sleep(Duration::from_secs(interval)).await;
+            }
+            Err((e, stats)) => {
+                emit_mev_earned_datapoint(stats);
+                datapoint_error!("mev-earned-error", ("error", e.to_string(), String),);
+                sleep(Duration::from_secs(5)).await;
+            }
+        };
+    }
+}
+
 async fn vote_account_loop(
     rpc_client: Arc<RpcClient>,
     keypair: Arc<Keypair>,
@@ -369,5 +406,12 @@ async fn main() {
         args.interval,
     ));
 
+    tokio::spawn(mev_earned_loop(
+        Arc::clone(&client),
+        Arc::clone(&keypair),
+        args.program_id,
+        args.tip_distribution_program_id,
+        args.interval,
+    ));
     gossip_upload_loop(client, keypair, args.program_id, entrypoint, args.interval).await;
 }

keepers/validator-keeper/src/mev_commission.rs

@@ -1,7 +1,8 @@
 use std::{collections::HashMap, str::FromStr, sync::Arc};
 
-use anchor_lang::{InstructionData, ToAccountMetas};
+use anchor_lang::{AccountDeserialize, InstructionData, ToAccountMetas};
 use jito_tip_distribution::sdk::derive_tip_distribution_account_address;
+use jito_tip_distribution::state::TipDistributionAccount;
 use keeper_core::{
     build_create_and_update_instructions, get_multiple_accounts_batched,
     get_vote_accounts_with_retry, submit_create_and_update, Address, CreateTransaction,
@@ -192,6 +193,73 @@ pub async fn update_mev_commission(
     submit_result.map_err(|(e, stats)| (e.into(), stats))
 }
 
+pub async fn update_mev_earned(
+    client: Arc<RpcClient>,
+    keypair: Arc<Keypair>,
+    validator_history_program_id: &Pubkey,
+    tip_distribution_program_id: &Pubkey,
+    validators_updated: &mut HashMap<Pubkey, Pubkey>,
+    curr_epoch: &mut u64,
+) -> Result<CreateUpdateStats, (MevCommissionError, CreateUpdateStats)> {
+    let epoch = client
+        .get_epoch_info()
+        .await
+        .map_err(|e| (e.into(), CreateUpdateStats::default()))?
+        .epoch;
+
+    if epoch > *curr_epoch {
+        // new epoch started, we assume here that all the validators with TDAs from curr_epoch-1 have had their merkle roots uploaded/processed by this point
+        // clear our map of TDAs derived from curr_epoch -1 and start fresh for epoch-1 (or curr_epoch)
+        validators_updated.clear();
+    }
+    *curr_epoch = epoch;
+
+    let vote_accounts = get_vote_accounts_with_retry(&client, MIN_VOTE_EPOCHS, None)
+        .await
+        .map_err(|e| (e.into(), CreateUpdateStats::default()))?;
+
+    let entries = vote_accounts
+        .iter()
+        .map(|vote_account| {
+            ValidatorMevCommissionEntry::new(
+                vote_account,
+                epoch - 1, // TDA derived from the prev epoch since the merkle roots are uploaded shortly after rollover
+                validator_history_program_id,
+                tip_distribution_program_id,
+                &keypair.pubkey(),
+            )
+        })
+        .collect::<Vec<ValidatorMevCommissionEntry>>();
+
+    let uploaded_merkleroot_entries =
+        get_entries_with_uploaded_merkleroot(client.clone(), &entries)
+            .await
+            .map_err(|e| (e.into(), CreateUpdateStats::default()))?;
+
+    let entries_to_update = uploaded_merkleroot_entries
+        .into_iter()
+        .filter(|entry| !validators_updated.contains_key(&entry.tip_distribution_account))
+        .collect::<Vec<ValidatorMevCommissionEntry>>();
+    let (create_transactions, update_instructions) =
+        build_create_and_update_instructions(&client, &entries_to_update)
+            .await
+            .map_err(|e| (e.into(), CreateUpdateStats::default()))?;
+
+    let submit_result =
+        submit_create_and_update(&client, create_transactions, update_instructions, &keypair).await;
+    if submit_result.is_ok() {
+        for ValidatorMevCommissionEntry {
+            vote_account,
+            tip_distribution_account,
+            ..
+        } in entries_to_update
+        {
+            validators_updated.insert(tip_distribution_account, vote_account);
+        }
+    }
+    submit_result.map_err(|(e, stats)| (e.into(), stats))
+}
+
 async fn get_existing_entries(
     client: Arc<RpcClient>,
     entries: &[ValidatorMevCommissionEntry],
@@ -217,3 +285,38 @@ async fn get_existing_entries(
     // Fetch existing tip distribution accounts for this epoch
     Ok(result)
 }
+
+async fn get_entries_with_uploaded_merkleroot(
+    client: Arc<RpcClient>,
+    entries: &[ValidatorMevCommissionEntry],
+) -> Result<Vec<ValidatorMevCommissionEntry>, MultipleAccountsError> {
+    /* Filters tip distribution tuples to the addresses, then fetches accounts to see which ones have an uploaded merkle root */
+    let tip_distribution_addresses = entries
+        .iter()
+        .map(|entry| entry.tip_distribution_account)
+        .collect::<Vec<Pubkey>>();
+
+    let accounts = get_multiple_accounts_batched(&tip_distribution_addresses, &client).await?;
+    let result = accounts
+        .iter()
+        .enumerate()
+        .filter_map(|(i, account_data)| {
+            if let Some(account_data) = account_data {
+                let mut data: &[u8] = &account_data.data;
+                if let Ok(tda) = TipDistributionAccount::try_deserialize(&mut data) {
+                    if let Some(_) = tda.merkle_root {
+                        return Some(entries[i].clone());
+                    } else {
+                        return None;
+                    }
+                } else {
+                    return None;
+                }
+            } else {
+                return None;
+            }
+        })
+        .collect::<Vec<ValidatorMevCommissionEntry>>();
+    // Fetch tip distribution accounts with uploaded merkle roots for this epoch
+    Ok(result)
+}

programs/validator-history/src/instructions/update_mev_commission.rs

@@ -1,11 +1,9 @@
-use anchor_lang::{
-    prelude::*,
-    solana_program::{clock::Clock, vote},
-};
+use anchor_lang::{prelude::*, solana_program::vote};
 
 use crate::{
     state::{Config, ValidatorHistory},
     utils::cast_epoch,
+    utils::fixed_point_sol,
 };
 use jito_tip_distribution::state::TipDistributionAccount;
 
@@ -56,10 +54,10 @@ pub fn handler(ctx: Context<UpdateMevCommission>, epoch: u64) -> Result<()> {
 
     let tip_distribution_account = TipDistributionAccount::try_deserialize(&mut tda_data)?;
     let mev_commission_bps = tip_distribution_account.validator_commission_bps;
-    let mut mev_earned: u64 = 0;
+    let mut mev_earned: u32 = 0;
     // if the merkle_root has been uploaded pull the mev_earned for the epoch
     if let Some(merkle_root) = tip_distribution_account.merkle_root {
-        mev_earned = merkle_root.max_total_claim;
+        mev_earned = fixed_point_sol(merkle_root.max_total_claim);
     }
     let epoch = cast_epoch(epoch);
     validator_history_account.set_mev_commission(epoch, mev_commission_bps, mev_earned)?;