API, Writer Service: Add directed stake target

Cargo.toml

@@ -39,7 +39,7 @@ http = { version = "0.2.1" }
 itertools = "0.5.9"
 jito-priority-fee-distribution = { features = ["no-entrypoint"], git = "https://github.com/jito-foundation/jito-programs", branch = "master" }
 jito-priority-fee-distribution-sdk = { git = "https://github.com/jito-foundation/jito-programs", branch = "master" }
-jito-steward = { features = ["no-entrypoint"], git = "https://github.com/jito-foundation/stakenet", rev = "d724a42b49c0cd32791afbd2a362dd63364a7a9a" }
+jito-steward = { package = "jito-steward", git = "https://github.com/jito-foundation/stakenet", rev = "6d35e58013cf7442dce287a1f45013085cb24967", features = ["no-entrypoint"] }
 jito-tip-distribution = { features = ["no-entrypoint"], git = "https://github.com/jito-foundation/jito-programs", branch = "master" }
 jito-tip-distribution-sdk = { git = "https://github.com/jito-foundation/jito-programs", branch = "master" }
 kobe-core = { path = "core", version = "1.0.0" }
@@ -83,7 +83,7 @@ spl-pod = "0.5.1"
 spl-stake-pool = { features = ["no-entrypoint"], version = "2.0.2" }
 spl-stake-pool-cli = { path = "cranker/cli", version = "1.0.0" }
 spl-token = { version = "4.0", features = ["no-entrypoint"] }
-stakenet-sdk = { package = "stakenet-sdk", git = "https://github.com/jito-foundation/stakenet", rev = "d724a42b49c0cd32791afbd2a362dd63364a7a9a" }
+stakenet-sdk = { package = "stakenet-sdk", git = "https://github.com/jito-foundation/stakenet", rev = "6d35e58013cf7442dce287a1f45013085cb24967" }
 thiserror = "1.0.37"
 tokio = { version = "1.36.0", features = ["full"] }
 tokio-retry = "0.3.0"
@@ -92,4 +92,4 @@ tower-http = { version = "0.4.0", features = ["cors", "trace"] }
 tracing = { version = "0.1.37" }
 tracing-core = "0.1.32"
 tracing-subscriber = { version = "0.3.17", features = ["env-filter"] }
-validator-history = { package = "validator-history", git = "https://github.com/jito-foundation/stakenet", rev = "d724a42b49c0cd32791afbd2a362dd63364a7a9a", features = ["no-entrypoint"] }
+validator-history = { package = "validator-history", git = "https://github.com/jito-foundation/stakenet", rev = "6d35e58013cf7442dce287a1f45013085cb24967", features = ["no-entrypoint"] }

README.md

@@ -310,7 +310,13 @@ RUST_LOG=info cargo r -p kobe-cranker -- \
 #### Start Data Collection
 
 ```bash
-cargo r --bin kobe-writer-service -- live
+RUST_LOG=info cargo r --bin kobe-writer-service -- \
+   --rpc-url "" \
+   --mongo-connection-uri "mongodb://localhost:27017/kobe" \
+   --mongo-db-name "validators" \
+   --solana-cluster "testnet" \
+   --steward-config-pubkey "5pZmpk3ktweGZW9xFknpEHhQoWeAKTzSGwnCUyVdiye" \
+   live
 ```
 
 #### Start Steward Monitoring

api/src/resolvers/query_resolver.rs

@@ -541,6 +541,7 @@ impl QueryResolver {
                         priority_fee_rewards: v.priority_fee_revenue_lamports,
                         running_jito: v.running_jito,
                         running_bam: v.running_bam,
+                        jito_directed_stake_target: v.jito_directed_stake_target,
                         jito_sol_active_lamports: None,
                     }
                 })
@@ -621,6 +622,7 @@ impl QueryResolver {
                         priority_fee_rewards: v.priority_fee_revenue_lamports,
                         running_jito: v.running_jito,
                         running_bam: v.running_bam,
+                        jito_directed_stake_target: v.jito_directed_stake_target,
                         jito_sol_active_lamports: Some(v.target_pool_active_lamports),
                     }
                 })

api/src/schemas/validator.rs

@@ -32,6 +32,9 @@ pub struct ValidatorEntry {
     /// Total active stake delegated to this validator on the Solana network
     pub active_stake: u64,
 
+    /// Whether or not jito directed stake target validator
+    pub jito_directed_stake_target: Option<bool>,
+
     /// Active stake lamports delegated to this validator from the JitoSOL stake-pool
     #[serde(skip_serializing_if = "Option::is_none")]
     pub jito_sol_active_lamports: Option<u64>,

core/src/db_models/validators.rs

@@ -27,6 +27,10 @@ pub struct Validator {
     pub epoch: u64,
     pub epoch_credits: Option<u64>,
     pub identity_account: Option<String>,
+
+    /// Whether or not jito directed stake target validator
+    pub jito_directed_stake_target: Option<bool>,
+
     pub mev_commission_bps: Option<u16>,
     pub mev_revenue_lamports: u64,
     pub priority_fee_commission_bps: Option<u16>,
@@ -82,6 +86,7 @@ impl Validator {
             epoch: validators_app_entry.epoch.unwrap_or_default(),
             epoch_credits: validators_app_entry.epoch_credits,
             identity_account: validators_app_entry.account.clone(),
+            jito_directed_stake_target: Some(on_chain_data.jito_directed_stake_target),
             mev_commission_bps: on_chain_data.mev_commission_bps,
             mev_revenue_lamports: on_chain_data.mev_revenue_lamports,
             priority_fee_commission_bps: Some(on_chain_data.priority_fee_commission_bps),

core/src/fetcher.rs

@@ -1,6 +1,7 @@
 use std::{
     collections::{HashMap, HashSet},
     str::FromStr,
+    sync::Arc,
 };
 
 use anchor_lang::AccountDeserialize;
@@ -12,7 +13,7 @@ use solana_client::{nonblocking::rpc_client::RpcClient, rpc_response::RpcVoteAcc
 use solana_pubkey::Pubkey;
 use spl_stake_pool::state::ValidatorStakeInfo;
 use spl_stake_pool_cli::client::get_validator_list;
-use stakenet_sdk::utils::accounts::get_all_validator_history_accounts;
+use stakenet_sdk::utils::accounts::{get_all_validator_history_accounts, get_directed_stake_meta};
 use validator_history::ValidatorHistory;
 
 use crate::{
@@ -65,6 +66,9 @@ pub struct ChainData {
     pub inflation_rewards_lamports: u64,
     pub priority_fee_commission_bps: u16,
     pub priority_fee_revenue_lamports: u64,
+
+    /// Jito Directed Stake Target
+    pub jito_directed_stake_target: bool,
 }
 
 pub fn get_tip_distribution_program_id(cluster: &Cluster) -> Pubkey {
@@ -125,22 +129,23 @@ pub fn get_priority_fee_distribution_program_id() -> solana_pubkey::Pubkey {
 pub async fn fetch_chain_data(
     validators: &[ValidatorsAppResponseEntry],
     bam_validator_set: HashSet<String>,
-    rpc_client: &RpcClient,
+    rpc_client: Arc<RpcClient>,
     cluster: &Cluster,
     epoch: u64,
     validator_list_pubkey: &Pubkey,
+    steward_config_pubkey: &Pubkey,
 ) -> Result<HashMap<Pubkey, ChainData>, Error> {
     // Fetch on-chain data
     let tip_distributions =
-        fetch_tip_distribution_accounts(validators, rpc_client, cluster, epoch).await?;
+        fetch_tip_distribution_accounts(validators, &rpc_client, cluster, epoch).await?;
     let priority_fee_distributions =
-        fetch_priority_fee_distribution_accounts(validators, rpc_client, epoch).await?;
+        fetch_priority_fee_distribution_accounts(validators, &rpc_client, epoch).await?;
     let vote_accounts = rpc_client.get_vote_accounts().await?;
     let (global_average, vote_credits_map) = fetch_vote_credits(&vote_accounts)?;
 
     let total_staked_lamports = fetch_total_staked_lamports(&vote_accounts);
 
-    let staked_validators = get_validator_list(rpc_client, validator_list_pubkey).await?;
+    let staked_validators = get_validator_list(&rpc_client, validator_list_pubkey).await?;
     let inflation_rate = match rpc_client.get_inflation_rate().await {
         Ok(rate) => rate.total,
         Err(e) => {
@@ -151,7 +156,14 @@ pub async fn fetch_chain_data(
 
     let validator_history_program_id = get_validator_history_program_id(cluster);
     let validator_histories =
-        fetch_validator_history_accounts(rpc_client, validator_history_program_id).await?;
+        fetch_validator_history_accounts(&rpc_client, validator_history_program_id).await?;
+
+    let directed_stake_meta = get_directed_stake_meta(
+        rpc_client.clone(),
+        steward_config_pubkey,
+        &jito_steward::id(),
+    )
+    .await?;
 
     Ok(HashMap::from_iter(validators.iter().map(|v| {
         let vote_account = v.vote_account;
@@ -217,6 +229,11 @@ pub async fn fetch_chain_data(
         let inflation_rewards_lamports =
             inflation_rate / epochs_per_year * staked_amount * vote_credit_proportion;
 
+        let jito_directed_stake_target = directed_stake_meta
+            .targets
+            .iter()
+            .any(|target| target.vote_pubkey.eq(&v.vote_account));
+
         let data = ChainData {
             mev_commission_bps,
             mev_revenue_lamports,
@@ -228,6 +245,7 @@ pub async fn fetch_chain_data(
             inflation_rewards_lamports: inflation_rewards_lamports as u64,
             priority_fee_commission_bps,
             priority_fee_revenue_lamports,
+            jito_directed_stake_target,
         };
 
         (vote_account, data)

writer-service/src/lib.rs

@@ -59,13 +59,15 @@ impl KobeWriterService {
     ///
     /// [`KobeWriterService`] uses multiple managers to fetch information such as validators, stake
     /// pool. so initialize these services in this function.
+    #[allow(clippy::too_many_arguments)]
     pub async fn new(
         mongo_connection_uri: &str,
         cluster: Cluster,
         rpc_url: String,
         tip_distribution_program_id: String,
         priority_fee_distribution_program_id: String,
         mainnet_gcp_server_names: Vec<String>,
+        steward_config: Pubkey,
         bam_api_base_url: Option<String>,
     ) -> Result<Self> {
         let mongodb_client = db::setup_mongo_client(mongo_connection_uri).await?;
@@ -89,8 +91,13 @@ impl KobeWriterService {
         .await
         .expect("Failed to initialize Validators App client");
 
-        let stake_pool_manager =
-            StakePoolManager::new(rpc_client, validators_app_client, bam_api_base_url, cluster);
+        let stake_pool_manager = StakePoolManager::new(
+            rpc_client,
+            validators_app_client,
+            bam_api_base_url,
+            cluster,
+            steward_config,
+        );
 
         Ok(Self {
             db,

writer-service/src/main.rs

@@ -5,6 +5,7 @@ use kobe_writer_service::{result::Result, KobeWriterService};
 use log::{error, info, set_boxed_logger, set_max_level, LevelFilter};
 use solana_clap_utils::input_validators::is_url_or_moniker;
 use solana_metrics::set_host_id;
+use solana_sdk::pubkey::Pubkey;
 use tokio::runtime::Runtime;
 
 #[derive(Parser)]
@@ -58,6 +59,14 @@ struct Args {
     )]
     priority_fee_distribution_program_id: String,
 
+    /// Steward config pubkey
+    #[arg(
+        long,
+        env,
+        default_value = "jitoVjT9jRUyeXHzvCwzPgHj7yWNRhLcUoXtes4wtjv"
+    )]
+    steward_config_pubkey: Pubkey,
+
     /// Mainnet gcp server names
     #[arg(long, env, value_delimiter = ',')]
     mainnet_gcp_server_names: Vec<String>,
@@ -111,6 +120,7 @@ fn main() -> Result<()> {
             args.tip_distribution_program_id,
             args.priority_fee_distribution_program_id,
             args.mainnet_gcp_server_names,
+            args.steward_config_pubkey,
             args.bam_api_base_url,
         )
         .await

writer-service/src/stake_pool_manager.rs

@@ -1,4 +1,6 @@
-use std::{collections::HashSet, str::FromStr, thread::sleep, time::Duration as CoreDuration};
+use std::{
+    collections::HashSet, str::FromStr, sync::Arc, thread::sleep, time::Duration as CoreDuration,
+};
 
 use bam_api_client::client::BamApiClient;
 use chrono::{Duration, DurationRound, Utc};
@@ -20,7 +22,7 @@ use crate::{result::Result, rpc_utils};
 
 pub struct StakePoolManager {
     /// RPC client
-    pub rpc_client: RpcClient,
+    pub rpc_client: Arc<RpcClient>,
 
     /// Validators app client
     pub validators_app_client: Client,
@@ -30,6 +32,9 @@ pub struct StakePoolManager {
 
     /// Cluster [Mainnet, Testnet, Devnet]
     pub cluster: Cluster,
+
+    /// Steward config pubkey
+    steward_config: Pubkey,
 }
 
 impl StakePoolManager {
@@ -38,12 +43,14 @@ impl StakePoolManager {
         validators_app_client: Client,
         bam_api_base_url: Option<String>,
         cluster: Cluster,
+        steward_config: Pubkey,
     ) -> Self {
         let mut manager = Self {
-            rpc_client,
+            rpc_client: Arc::new(rpc_client),
             validators_app_client,
             bam_api_client: None,
             cluster,
+            steward_config,
         };
 
         if let Some(bam_api_base_url) = bam_api_base_url {
@@ -80,10 +87,11 @@ impl StakePoolManager {
         let on_chain_data = fetch_chain_data(
             network_validators.as_ref(),
             bam_validator_set,
-            &self.rpc_client,
+            self.rpc_client.clone(),
             &self.cluster,
             epoch,
             validator_list_address,
+            &self.steward_config,
         )
         .await?;