Add validator catchup command (#6922)

cli/Cargo.toml

@@ -19,6 +19,7 @@ console = "0.9.1"
 dirs = "2.0.2"
 lazy_static = "1.4.0"
 log = "0.4.8"
+indicatif = "0.13.0"
 num-traits = "0.2"
 pretty-hex = "0.1.1"
 reqwest = { version = "0.9.22", default-features = false, features = ["rustls-tls"] }

cli/src/cli.rs

@@ -6,7 +6,6 @@ use clap::{App, AppSettings, Arg, ArgMatches, SubCommand};
 use log::*;
 use num_traits::FromPrimitive;
 use serde_json::{self, json, Value};
-
 use solana_budget_api::budget_instruction::{self, BudgetError};
 use solana_clap_utils::{input_parsers::*, input_validators::*};
 use solana_client::{client_error::ClientError, rpc_client::RpcClient};
@@ -33,7 +32,6 @@ use solana_sdk::{
 use solana_stake_api::stake_state::{Lockup, StakeAuthorize};
 use solana_storage_api::storage_instruction::StorageAccountType;
 use solana_vote_api::vote_state::VoteAuthorize;
-
 use std::{
     fs::File,
     io::{Read, Write},
@@ -71,6 +69,9 @@ impl std::ops::Deref for KeypairEq {
 #[allow(clippy::large_enum_variant)]
 pub enum CliCommand {
     // Cluster Query Commands
+    Catchup {
+        node_pubkey: Pubkey,
+    },
     ClusterVersion,
     Fees,
     GetEpochInfo,
@@ -237,6 +238,7 @@ impl Default for CliConfig {
 pub fn parse_command(matches: &ArgMatches<'_>) -> Result<CliCommandInfo, Box<dyn error::Error>> {
     let response = match matches.subcommand() {
         // Cluster Query Commands
+        ("catchup", Some(matches)) => parse_catchup(matches),
         ("cluster-version", Some(_matches)) => Ok(CliCommandInfo {
             command: CliCommand::ClusterVersion,
             require_keypair: false,
@@ -849,7 +851,8 @@ pub fn process_command(config: &CliConfig) -> ProcessResult {
         // Cluster Query Commands
 
         // Return software version of solana-cli and cluster entrypoint node
-        CliCommand::ClusterVersion => process_cluster_version(&rpc_client, config),
+        CliCommand::Catchup { node_pubkey } => process_catchup(&rpc_client, node_pubkey),
+        CliCommand::ClusterVersion => process_cluster_version(&rpc_client),
         CliCommand::Fees => process_fees(&rpc_client),
         CliCommand::GetGenesisHash => process_get_genesis_hash(&rpc_client),
         CliCommand::GetSlot => process_get_slot(&rpc_client),

cli/src/cluster_query.rs

@@ -7,16 +7,20 @@ use crate::{
 };
 use clap::{value_t_or_exit, App, Arg, ArgMatches, SubCommand};
 use console::{style, Emoji};
+use indicatif::{ProgressBar, ProgressStyle};
+use solana_clap_utils::{input_parsers::*, input_validators::*};
 use solana_client::{rpc_client::RpcClient, rpc_request::RpcVoteAccountInfo};
 use solana_sdk::{
     clock,
     commitment_config::CommitmentConfig,
     hash::Hash,
+    pubkey::Pubkey,
     signature::{Keypair, KeypairUtil},
     system_transaction,
 };
 use std::{
     collections::VecDeque,
+    thread::sleep,
     time::{Duration, Instant},
 };
 
@@ -31,6 +35,19 @@ pub trait ClusterQuerySubCommands {
 impl ClusterQuerySubCommands for App<'_, '_> {
     fn cluster_query_subcommands(self) -> Self {
         self.subcommand(
+            SubCommand::with_name("catchup")
+                .about("Wait for a validator to catch up to the cluster")
+                .arg(
+                    Arg::with_name("node_pubkey")
+                        .index(1)
+                        .takes_value(true)
+                        .value_name("PUBKEY")
+                        .validator(is_pubkey_or_keypair)
+                        .required(true)
+                        .help("Identity pubkey of the validator"),
+                ),
+        )
+        .subcommand(
             SubCommand::with_name("cluster-version")
                 .about("Get the version of the cluster entrypoint"),
         )
@@ -97,6 +114,14 @@ impl ClusterQuerySubCommands for App<'_, '_> {
     }
 }
 
+pub fn parse_catchup(matches: &ArgMatches<'_>) -> Result<CliCommandInfo, CliError> {
+    let node_pubkey = pubkey_of(matches, "node_pubkey").unwrap();
+    Ok(CliCommandInfo {
+        command: CliCommand::Catchup { node_pubkey },
+        require_keypair: false,
+    })
+}
+
 pub fn parse_cluster_ping(matches: &ArgMatches<'_>) -> Result<CliCommandInfo, CliError> {
     let interval = Duration::from_secs(value_t_or_exit!(matches, "interval", u64));
     let count = if matches.is_present("count") {
@@ -130,7 +155,74 @@ pub fn parse_show_validators(matches: &ArgMatches<'_>) -> Result<CliCommandInfo,
     })
 }
 
-pub fn process_cluster_version(rpc_client: &RpcClient, _config: &CliConfig) -> ProcessResult {
+/// Creates a new process bar for processing that will take an unknown amount of time
+fn new_spinner_progress_bar() -> ProgressBar {
+    let progress_bar = ProgressBar::new(42);
+    progress_bar
+        .set_style(ProgressStyle::default_spinner().template("{spinner:.green} {wide_msg}"));
+    progress_bar.enable_steady_tick(100);
+    progress_bar
+}
+
+pub fn process_catchup(rpc_client: &RpcClient, node_pubkey: &Pubkey) -> ProcessResult {
+    let cluster_nodes = rpc_client.get_cluster_nodes()?;
+
+    let rpc_addr = cluster_nodes
+        .iter()
+        .find(|contact_info| contact_info.pubkey == node_pubkey.to_string())
+        .ok_or_else(|| format!("Contact information not found for {}", node_pubkey))?
+        .rpc
+        .ok_or_else(|| format!("RPC service not found for {}", node_pubkey))?;
+
+    let progress_bar = new_spinner_progress_bar();
+    progress_bar.set_message("Connecting...");
+
+    let node_client = RpcClient::new_socket(rpc_addr);
+    let mut previous_rpc_slot = std::u64::MAX;
+    let mut previous_slot_distance = 0;
+    let sleep_interval = 5;
+    loop {
+        let rpc_slot = rpc_client.get_slot_with_commitment(CommitmentConfig::recent())?;
+        let node_slot = node_client.get_slot_with_commitment(CommitmentConfig::recent())?;
+        if node_slot > std::cmp::min(previous_rpc_slot, rpc_slot) {
+            progress_bar.finish_and_clear();
+            return Ok(format!(
+                "{} has caught up (us:{} them:{})",
+                node_pubkey, node_slot, rpc_slot,
+            ));
+        }
+
+        let slot_distance = rpc_slot as i64 - node_slot as i64;
+        progress_bar.set_message(&format!(
+            "Validator is {} slots away (us:{} them:{}){}",
+            slot_distance,
+            node_slot,
+            rpc_slot,
+            if previous_rpc_slot == std::u64::MAX {
+                "".to_string()
+            } else {
+                let slots_per_second =
+                    (previous_slot_distance - slot_distance) as f64 / f64::from(sleep_interval);
+
+                format!(
+                    " and {} at {:.1} slots/second",
+                    if slots_per_second < 0.0 {
+                        "falling behind"
+                    } else {
+                        "gaining"
+                    },
+                    slots_per_second,
+                )
+            }
+        ));
+
+        sleep(Duration::from_secs(sleep_interval as u64));
+        previous_rpc_slot = rpc_slot;
+        previous_slot_distance = slot_distance;
+    }
+}
+
+pub fn process_cluster_version(rpc_client: &RpcClient) -> ProcessResult {
     let remote_version = rpc_client.get_version()?;
     Ok(remote_version.solana_core)
 }