use {
    log::*,
    rand::{seq::SliceRandom, thread_rng, Rng},
    rayon::prelude::*,
    solana_core::validator::{ValidatorConfig, ValidatorStartProgress},
    solana_download_utils::{download_snapshot_archive, DownloadProgressRecord},
    solana_genesis_utils::download_then_check_genesis_hash,
    solana_gossip::{
        cluster_info::{ClusterInfo, Node},
        crds_value,
        gossip_service::GossipService,
        legacy_contact_info::LegacyContactInfo as ContactInfo,
    },
    solana_metrics::datapoint_info,
    solana_rpc_client::rpc_client::RpcClient,
    solana_runtime::{
        snapshot_archive_info::SnapshotArchiveInfoGetter,
        snapshot_package::SnapshotType,
        snapshot_utils::{self},
    },
    solana_sdk::{
        clock::Slot,
        commitment_config::CommitmentConfig,
        hash::Hash,
        pubkey::Pubkey,
        signature::{Keypair, Signer},
    },
    solana_streamer::socket::SocketAddrSpace,
    std::{
        collections::{hash_map::RandomState, HashMap, HashSet},
        net::{SocketAddr, TcpListener, UdpSocket},
        path::Path,
        process::exit,
        sync::{
            atomic::{AtomicBool, Ordering},
            Arc, RwLock,
        },
        time::{Duration, Instant},
    },
    thiserror::Error,
};

/// When downloading snapshots, wait at most this long for snapshot hashes from
/// _all_ known validators.  Afterwards, wait for snapshot hashes from _any_
/// known validator.
const WAIT_FOR_ALL_KNOWN_VALIDATORS: Duration = Duration::from_secs(60);
/// If we don't have any alternative peers after this long, better off trying
/// blacklisted peers again.
const BLACKLIST_CLEAR_THRESHOLD: Duration = Duration::from_secs(60);
/// If we can't find a good snapshot download candidate after this time, just
/// give up.
const NEWER_SNAPSHOT_THRESHOLD: Duration = Duration::from_secs(180);
/// If we haven't found any RPC peers after this time, just give up.
const GET_RPC_PEERS_TIMEOUT: Duration = Duration::from_secs(300);

pub const MAX_RPC_CONNECTIONS_EVALUATED_PER_ITERATION: usize = 32;

#[derive(Debug)]
pub struct RpcBootstrapConfig {
    pub no_genesis_fetch: bool,
    pub no_snapshot_fetch: bool,
    pub only_known_rpc: bool,
    pub max_genesis_archive_unpacked_size: u64,
    pub check_vote_account: Option<String>,
    pub incremental_snapshot_fetch: bool,
}

fn verify_reachable_ports(
    node: &Node,
    cluster_entrypoint: &ContactInfo,
    validator_config: &ValidatorConfig,
    socket_addr_space: &SocketAddrSpace,
) -> bool {
    let verify_address = |addr: &Option<SocketAddr>| -> bool {
        addr.as_ref()
            .map(|addr| socket_addr_space.check(addr))
            .unwrap_or_default()
    };
    let mut udp_sockets = vec![&node.sockets.gossip, &node.sockets.repair];

    if verify_address(&node.info.serve_repair().ok()) {
        udp_sockets.push(&node.sockets.serve_repair);
    }
    if verify_address(&node.info.tpu().ok()) {
        udp_sockets.extend(node.sockets.tpu.iter());
        udp_sockets.push(&node.sockets.tpu_quic);
    }
    if verify_address(&node.info.tpu_forwards().ok()) {
        udp_sockets.extend(node.sockets.tpu_forwards.iter());
        udp_sockets.push(&node.sockets.tpu_forwards_quic);
    }
    if verify_address(&node.info.tpu_vote().ok()) {
        udp_sockets.extend(node.sockets.tpu_vote.iter());
    }
    if verify_address(&node.info.tvu().ok()) {
        udp_sockets.extend(node.sockets.tvu.iter());
        udp_sockets.extend(node.sockets.broadcast.iter());
        udp_sockets.extend(node.sockets.retransmit_sockets.iter());
    }
    if verify_address(&node.info.tvu_forwards().ok()) {
        udp_sockets.extend(node.sockets.tvu_forwards.iter());
    }

    let mut tcp_listeners = vec![];
    if let Some((rpc_addr, rpc_pubsub_addr)) = validator_config.rpc_addrs {
        for (purpose, bind_addr, public_addr) in &[
            ("RPC", rpc_addr, node.info.rpc()),
            ("RPC pubsub", rpc_pubsub_addr, node.info.rpc_pubsub()),
        ] {
            if verify_address(&public_addr.as_ref().ok().copied()) {
                tcp_listeners.push((
                    bind_addr.port(),
                    TcpListener::bind(bind_addr).unwrap_or_else(|err| {
                        error!(
                            "Unable to bind to tcp {:?} for {}: {}",
                            bind_addr, purpose, err
                        );
                        exit(1);
                    }),
                ));
            }
        }
    }

    if let Some(ip_echo) = &node.sockets.ip_echo {
        let ip_echo = ip_echo.try_clone().expect("unable to clone tcp_listener");
        tcp_listeners.push((ip_echo.local_addr().unwrap().port(), ip_echo));
    }

    solana_net_utils::verify_reachable_ports(
        &cluster_entrypoint.gossip,
        tcp_listeners,
        &udp_sockets,
    )
}

fn is_known_validator(id: &Pubkey, known_validators: &Option<HashSet<Pubkey>>) -> bool {
    if let Some(known_validators) = known_validators {
        known_validators.contains(id)
    } else {
        false
    }
}

fn start_gossip_node(
    identity_keypair: Arc<Keypair>,
    cluster_entrypoints: &[ContactInfo],
    ledger_path: &Path,
    gossip_addr: &SocketAddr,
    gossip_socket: UdpSocket,
    expected_shred_version: Option<u16>,
    gossip_validators: Option<HashSet<Pubkey>>,
    should_check_duplicate_instance: bool,
    socket_addr_space: SocketAddrSpace,
) -> (Arc<ClusterInfo>, Arc<AtomicBool>, GossipService) {
    let contact_info = ClusterInfo::gossip_contact_info(
        identity_keypair.pubkey(),
        *gossip_addr,
        expected_shred_version.unwrap_or(0),
    );
    let mut cluster_info = ClusterInfo::new(contact_info, identity_keypair, socket_addr_space);
    cluster_info.set_entrypoints(cluster_entrypoints.to_vec());
    cluster_info.restore_contact_info(ledger_path, 0);
    let cluster_info = Arc::new(cluster_info);

    let gossip_exit_flag = Arc::new(AtomicBool::new(false));
    let gossip_service = GossipService::new(
        &cluster_info,
        None,
        gossip_socket,
        gossip_validators,
        should_check_duplicate_instance,
        None,
        &gossip_exit_flag,
    );
    (cluster_info, gossip_exit_flag, gossip_service)
}

fn get_rpc_peers(
    cluster_info: &ClusterInfo,
    cluster_entrypoints: &[ContactInfo],
    validator_config: &ValidatorConfig,
    blacklisted_rpc_nodes: &mut HashSet<Pubkey>,
    blacklist_timeout: &Instant,
    retry_reason: &mut Option<String>,
    bootstrap_config: &RpcBootstrapConfig,
) -> Vec<ContactInfo> {
    let shred_version = validator_config
        .expected_shred_version
        .unwrap_or_else(|| cluster_info.my_shred_version());
    if shred_version == 0 {
        let all_zero_shred_versions = cluster_entrypoints.iter().all(|cluster_entrypoint| {
            cluster_info
                .lookup_contact_info_by_gossip_addr(&cluster_entrypoint.gossip)
                .map_or(false, |entrypoint| entrypoint.shred_version == 0)
        });

        if all_zero_shred_versions {
            eprintln!("Entrypoint shred version is zero.  Restart with --expected-shred-version");
            exit(1);
        }
        info!("Waiting to adopt entrypoint shred version...");
        return vec![];
    }

    info!(
        "Searching for an RPC service with shred version {}{}...",
        shred_version,
        retry_reason
            .as_ref()
            .map(|s| format!(" (Retrying: {s})"))
            .unwrap_or_default()
    );

    let mut rpc_peers = cluster_info
        .all_rpc_peers()
        .into_iter()
        .filter(|contact_info| contact_info.shred_version == shred_version)
        .collect::<Vec<_>>();

    if bootstrap_config.only_known_rpc {
        rpc_peers.retain(|rpc_peer| {
            is_known_validator(&rpc_peer.id, &validator_config.known_validators)
        });
    }

    let rpc_peers_total = rpc_peers.len();

    // Filter out blacklisted nodes
    let rpc_peers: Vec<_> = rpc_peers
        .into_iter()
        .filter(|rpc_peer| !blacklisted_rpc_nodes.contains(&rpc_peer.id))
        .collect();
    let rpc_peers_blacklisted = rpc_peers_total - rpc_peers.len();
    let rpc_known_peers = rpc_peers
        .iter()
        .filter(|rpc_peer| is_known_validator(&rpc_peer.id, &validator_config.known_validators))
        .count();

    info!(
        "Total {} RPC nodes found. {} known, {} blacklisted ",
        rpc_peers_total, rpc_known_peers, rpc_peers_blacklisted
    );

    if rpc_peers_blacklisted == rpc_peers_total {
        *retry_reason = if !blacklisted_rpc_nodes.is_empty()
            && blacklist_timeout.elapsed() > BLACKLIST_CLEAR_THRESHOLD
        {
            // All nodes are blacklisted and no additional nodes recently discovered.
            // Remove all nodes from the blacklist and try them again.
            blacklisted_rpc_nodes.clear();
            Some("Blacklist timeout expired".to_owned())
        } else {
            Some("Wait for known rpc peers".to_owned())
        };
        return vec![];
    }
    rpc_peers
}

fn check_vote_account(
    rpc_client: &RpcClient,
    identity_pubkey: &Pubkey,
    vote_account_address: &Pubkey,
    authorized_voter_pubkeys: &[Pubkey],
) -> Result<(), String> {
    let vote_account = rpc_client
        .get_account_with_commitment(vote_account_address, CommitmentConfig::confirmed())
        .map_err(|err| format!("failed to fetch vote account: {err}"))?
        .value
        .ok_or_else(|| format!("vote account does not exist: {vote_account_address}"))?;

    if vote_account.owner != solana_vote_program::id() {
        return Err(format!(
            "not a vote account (owned by {}): {}",
            vote_account.owner, vote_account_address
        ));
    }

    let identity_account = rpc_client
        .get_account_with_commitment(identity_pubkey, CommitmentConfig::confirmed())
        .map_err(|err| format!("failed to fetch identity account: {err}"))?
        .value
        .ok_or_else(|| format!("identity account does not exist: {identity_pubkey}"))?;

    let vote_state = solana_vote_program::vote_state::from(&vote_account);
    if let Some(vote_state) = vote_state {
        if vote_state.authorized_voters().is_empty() {
            return Err("Vote account not yet initialized".to_string());
        }

        if vote_state.node_pubkey != *identity_pubkey {
            return Err(format!(
                "vote account's identity ({}) does not match the validator's identity {}).",
                vote_state.node_pubkey, identity_pubkey
            ));
        }

        for (_, vote_account_authorized_voter_pubkey) in vote_state.authorized_voters().iter() {
            if !authorized_voter_pubkeys.contains(vote_account_authorized_voter_pubkey) {
                return Err(format!(
                    "authorized voter {vote_account_authorized_voter_pubkey} not available"
                ));
            }
        }
    } else {
        return Err(format!(
            "invalid vote account data for {vote_account_address}"
        ));
    }

    // Maybe we can calculate minimum voting fee; rather than 1 lamport
    if identity_account.lamports <= 1 {
        return Err(format!(
            "underfunded identity account ({}): only {} lamports available",
            identity_pubkey, identity_account.lamports
        ));
    }

    Ok(())
}

#[derive(Error, Debug)]
pub enum GetRpcNodeError {
    #[error("Unable to find any RPC peers")]
    NoRpcPeersFound,

    #[error("Giving up, did not get newer snapshots from the cluster")]
    NoNewerSnapshots,
}

/// Struct to wrap the return value from get_rpc_nodes().  The `rpc_contact_info` is the peer to
/// download from, and `snapshot_hash` is the (optional) full and (optional) incremental
/// snapshots to download.
#[derive(Debug)]
struct GetRpcNodeResult {
    rpc_contact_info: ContactInfo,
    snapshot_hash: Option<SnapshotHash>,
}

/// Struct to wrap the peers & snapshot hashes together.
#[derive(Debug, PartialEq, Eq, Clone)]
struct PeerSnapshotHash {
    rpc_contact_info: ContactInfo,
    snapshot_hash: SnapshotHash,
}

/// A snapshot hash.  In this context (bootstrap *with* incremental snapshots), a snapshot hash
/// is _both_ a full snapshot hash and an (optional) incremental snapshot hash.
#[derive(Debug, PartialEq, Eq, PartialOrd, Ord, Clone, Copy)]
pub struct SnapshotHash {
    full: (Slot, Hash),
    incr: Option<(Slot, Hash)>,
}

pub fn fail_rpc_node(
    err: String,
    known_validators: &Option<HashSet<Pubkey, RandomState>>,
    rpc_id: &Pubkey,
    blacklisted_rpc_nodes: &mut HashSet<Pubkey, RandomState>,
) {
    warn!("{}", err);
    if let Some(ref known_validators) = known_validators {
        if known_validators.contains(rpc_id) {
            return;
        }
    }

    info!("Excluding {} as a future RPC candidate", rpc_id);
    blacklisted_rpc_nodes.insert(*rpc_id);
}

fn shutdown_gossip_service(gossip: (Arc<ClusterInfo>, Arc<AtomicBool>, GossipService)) {
    let (cluster_info, gossip_exit_flag, gossip_service) = gossip;
    cluster_info.save_contact_info();
    gossip_exit_flag.store(true, Ordering::Relaxed);
    gossip_service.join().unwrap();
}

#[allow(clippy::too_many_arguments)]
pub fn attempt_download_genesis_and_snapshot(
    rpc_contact_info: &ContactInfo,
    ledger_path: &Path,
    validator_config: &mut ValidatorConfig,
    bootstrap_config: &RpcBootstrapConfig,
    use_progress_bar: bool,
    gossip: &mut Option<(Arc<ClusterInfo>, Arc<AtomicBool>, GossipService)>,
    rpc_client: &RpcClient,
    full_snapshot_archives_dir: &Path,
    incremental_snapshot_archives_dir: &Path,
    maximum_local_snapshot_age: Slot,
    start_progress: &Arc<RwLock<ValidatorStartProgress>>,
    minimal_snapshot_download_speed: f32,
    maximum_snapshot_download_abort: u64,
    download_abort_count: &mut u64,
    snapshot_hash: Option<SnapshotHash>,
    identity_keypair: &Arc<Keypair>,
    vote_account: &Pubkey,
    authorized_voter_keypairs: Arc<RwLock<Vec<Arc<Keypair>>>>,
) -> Result<(), String> {
    download_then_check_genesis_hash(
        &rpc_contact_info.rpc,
        ledger_path,
        &mut validator_config.expected_genesis_hash,
        bootstrap_config.max_genesis_archive_unpacked_size,
        bootstrap_config.no_genesis_fetch,
        use_progress_bar,
        rpc_client,
    )?;

    if let Some(gossip) = gossip.take() {
        shutdown_gossip_service(gossip);
    }

    let rpc_client_slot = rpc_client
        .get_slot_with_commitment(CommitmentConfig::finalized())
        .map_err(|err| format!("Failed to get RPC node slot: {err}"))?;
    info!("RPC node root slot: {}", rpc_client_slot);

    download_snapshots(
        full_snapshot_archives_dir,
        incremental_snapshot_archives_dir,
        validator_config,
        bootstrap_config,
        use_progress_bar,
        maximum_local_snapshot_age,
        start_progress,
        minimal_snapshot_download_speed,
        maximum_snapshot_download_abort,
        download_abort_count,
        snapshot_hash,
        rpc_contact_info,
    )?;

    if let Some(url) = bootstrap_config.check_vote_account.as_ref() {
        let rpc_client = RpcClient::new(url);
        check_vote_account(
            &rpc_client,
            &identity_keypair.pubkey(),
            vote_account,
            &authorized_voter_keypairs
                .read()
                .unwrap()
                .iter()
                .map(|k| k.pubkey())
                .collect::<Vec<_>>(),
        )
        .unwrap_or_else(|err| {
            // Consider failures here to be more likely due to user error (eg,
            // incorrect `solana-validator` command-line arguments) rather than the
            // RPC node failing.
            //
            // Power users can always use the `--no-check-vote-account` option to
            // bypass this check entirely
            error!("{}", err);
            exit(1);
        });
    }
    Ok(())
}

// Populates `vetted_rpc_nodes` with a list of RPC nodes that are ready to be
// used for downloading latest snapshots and/or the genesis block. Guaranteed to
// find at least one viable node or terminate the process.
fn get_vetted_rpc_nodes(
    vetted_rpc_nodes: &mut Vec<(ContactInfo, Option<SnapshotHash>, RpcClient)>,
    cluster_info: &Arc<ClusterInfo>,
    cluster_entrypoints: &[ContactInfo],
    validator_config: &ValidatorConfig,
    blacklisted_rpc_nodes: &mut HashSet<Pubkey>,
    bootstrap_config: &RpcBootstrapConfig,
) {
    while vetted_rpc_nodes.is_empty() {
        let rpc_node_details = match get_rpc_nodes(
            cluster_info,
            cluster_entrypoints,
            validator_config,
            blacklisted_rpc_nodes,
            bootstrap_config,
        ) {
            Ok(rpc_node_details) => rpc_node_details,
            Err(err) => {
                error!(
                    "Failed to get RPC nodes: {err}. Consider checking system \
                    clock, removing `--no-port-check`, or adjusting \
                    `--known-validator ...` arguments as applicable"
                );
                exit(1);
            }
        };

        let newly_blacklisted_rpc_nodes = RwLock::new(HashSet::new());
        vetted_rpc_nodes.extend(
            rpc_node_details
                .into_par_iter()
                .map(|rpc_node_details| {
                    let GetRpcNodeResult {
                        rpc_contact_info,
                        snapshot_hash,
                    } = rpc_node_details;

                    info!(
                        "Using RPC service from node {}: {:?}",
                        rpc_contact_info.id, rpc_contact_info.rpc
                    );
                    let rpc_client = RpcClient::new_socket_with_timeout(
                        rpc_contact_info.rpc,
                        Duration::from_secs(5),
                    );

                    (rpc_contact_info, snapshot_hash, rpc_client)
                })
                .filter(|(rpc_contact_info, _snapshot_hash, rpc_client)| {
                    match rpc_client.get_version() {
                        Ok(rpc_version) => {
                            info!("RPC node version: {}", rpc_version.solana_core);
                            true
                        }
                        Err(err) => {
                            fail_rpc_node(
                                format!("Failed to get RPC node version: {err}"),
                                &validator_config.known_validators,
                                &rpc_contact_info.id,
                                &mut newly_blacklisted_rpc_nodes.write().unwrap(),
                            );
                            false
                        }
                    }
                })
                .collect::<Vec<(ContactInfo, Option<SnapshotHash>, RpcClient)>>(),
        );
        blacklisted_rpc_nodes.extend(newly_blacklisted_rpc_nodes.into_inner().unwrap());
    }
}

#[allow(clippy::too_many_arguments)]
pub fn rpc_bootstrap(
    node: &Node,
    identity_keypair: &Arc<Keypair>,
    ledger_path: &Path,
    full_snapshot_archives_dir: &Path,
    incremental_snapshot_archives_dir: &Path,
    vote_account: &Pubkey,
    authorized_voter_keypairs: Arc<RwLock<Vec<Arc<Keypair>>>>,
    cluster_entrypoints: &[ContactInfo],
    validator_config: &mut ValidatorConfig,
    bootstrap_config: RpcBootstrapConfig,
    do_port_check: bool,
    use_progress_bar: bool,
    maximum_local_snapshot_age: Slot,
    should_check_duplicate_instance: bool,
    start_progress: &Arc<RwLock<ValidatorStartProgress>>,
    minimal_snapshot_download_speed: f32,
    maximum_snapshot_download_abort: u64,
    socket_addr_space: SocketAddrSpace,
) {
    if do_port_check {
        let mut order: Vec<_> = (0..cluster_entrypoints.len()).collect();
        order.shuffle(&mut thread_rng());
        if order.into_iter().all(|i| {
            !verify_reachable_ports(
                node,
                &cluster_entrypoints[i],
                validator_config,
                &socket_addr_space,
            )
        }) {
            exit(1);
        }
    }

    if bootstrap_config.no_genesis_fetch && bootstrap_config.no_snapshot_fetch {
        return;
    }

    let total_snapshot_download_time = Instant::now();
    let mut get_rpc_nodes_time = Duration::new(0, 0);
    let mut snapshot_download_time = Duration::new(0, 0);
    let mut blacklisted_rpc_nodes = HashSet::new();
    let mut gossip = None;
    let mut vetted_rpc_nodes = vec![];
    let mut download_abort_count = 0;
    loop {
        if gossip.is_none() {
            *start_progress.write().unwrap() = ValidatorStartProgress::SearchingForRpcService;

            gossip = Some(start_gossip_node(
                identity_keypair.clone(),
                cluster_entrypoints,
                ledger_path,
                &node
                    .info
                    .gossip()
                    .expect("Operator must spin up node with valid gossip address"),
                node.sockets.gossip.try_clone().unwrap(),
                validator_config.expected_shred_version,
                validator_config.gossip_validators.clone(),
                should_check_duplicate_instance,
                socket_addr_space,
            ));
        }

        let get_rpc_nodes_start = Instant::now();
        get_vetted_rpc_nodes(
            &mut vetted_rpc_nodes,
            &gossip.as_ref().unwrap().0,
            cluster_entrypoints,
            validator_config,
            &mut blacklisted_rpc_nodes,
            &bootstrap_config,
        );
        let (rpc_contact_info, snapshot_hash, rpc_client) = vetted_rpc_nodes.pop().unwrap();
        get_rpc_nodes_time += get_rpc_nodes_start.elapsed();

        let snapshot_download_start = Instant::now();
        let download_result = attempt_download_genesis_and_snapshot(
            &rpc_contact_info,
            ledger_path,
            validator_config,
            &bootstrap_config,
            use_progress_bar,
            &mut gossip,
            &rpc_client,
            full_snapshot_archives_dir,
            incremental_snapshot_archives_dir,
            maximum_local_snapshot_age,
            start_progress,
            minimal_snapshot_download_speed,
            maximum_snapshot_download_abort,
            &mut download_abort_count,
            snapshot_hash,
            identity_keypair,
            vote_account,
            authorized_voter_keypairs.clone(),
        );
        snapshot_download_time += snapshot_download_start.elapsed();
        match download_result {
            Ok(()) => break,
            Err(err) => {
                fail_rpc_node(
                    err,
                    &validator_config.known_validators,
                    &rpc_contact_info.id,
                    &mut blacklisted_rpc_nodes,
                );
            }
        }
    }

    if let Some(gossip) = gossip.take() {
        shutdown_gossip_service(gossip);
    }

    datapoint_info!(
        "bootstrap-snapshot-download",
        (
            "total_time_secs",
            total_snapshot_download_time.elapsed().as_secs(),
            i64
        ),
        ("get_rpc_nodes_time_secs", get_rpc_nodes_time.as_secs(), i64),
        (
            "snapshot_download_time_secs",
            snapshot_download_time.as_secs(),
            i64
        ),
        ("download_abort_count", download_abort_count, i64),
        ("blacklisted_nodes_count", blacklisted_rpc_nodes.len(), i64),
    );
}

/// Get RPC peer node candidates to download from.
///
/// This function finds the highest compatible snapshots from the cluster and returns RPC peers.
fn get_rpc_nodes(
    cluster_info: &ClusterInfo,
    cluster_entrypoints: &[ContactInfo],
    validator_config: &ValidatorConfig,
    blacklisted_rpc_nodes: &mut HashSet<Pubkey>,
    bootstrap_config: &RpcBootstrapConfig,
) -> Result<Vec<GetRpcNodeResult>, GetRpcNodeError> {
    let mut blacklist_timeout = Instant::now();
    let mut get_rpc_peers_timout = Instant::now();
    let mut newer_cluster_snapshot_timeout = None;
    let mut retry_reason = None;
    loop {
        // Give gossip some time to populate and not spin on grabbing the crds lock
        std::thread::sleep(Duration::from_secs(1));
        info!("\n{}", cluster_info.rpc_info_trace());

        let rpc_peers = get_rpc_peers(
            cluster_info,
            cluster_entrypoints,
            validator_config,
            blacklisted_rpc_nodes,
            &blacklist_timeout,
            &mut retry_reason,
            bootstrap_config,
        );
        if rpc_peers.is_empty() {
            if get_rpc_peers_timout.elapsed() > GET_RPC_PEERS_TIMEOUT {
                return Err(GetRpcNodeError::NoRpcPeersFound);
            }
            continue;
        }

        // Reset timeouts if we found any viable RPC peers.
        blacklist_timeout = Instant::now();
        get_rpc_peers_timout = Instant::now();
        if bootstrap_config.no_snapshot_fetch {
            let random_peer = &rpc_peers[thread_rng().gen_range(0, rpc_peers.len())];
            return Ok(vec![GetRpcNodeResult {
                rpc_contact_info: random_peer.clone(),
                snapshot_hash: None,
            }]);
        }

        let known_validators_to_wait_for = if newer_cluster_snapshot_timeout
            .as_ref()
            .map(|timer: &Instant| timer.elapsed() < WAIT_FOR_ALL_KNOWN_VALIDATORS)
            .unwrap_or(true)
        {
            KnownValidatorsToWaitFor::All
        } else {
            KnownValidatorsToWaitFor::Any
        };
        let peer_snapshot_hashes = get_peer_snapshot_hashes(
            cluster_info,
            &rpc_peers,
            validator_config.known_validators.as_ref(),
            known_validators_to_wait_for,
            bootstrap_config.incremental_snapshot_fetch,
        );
        if peer_snapshot_hashes.is_empty() {
            match newer_cluster_snapshot_timeout {
                None => newer_cluster_snapshot_timeout = Some(Instant::now()),
                Some(newer_cluster_snapshot_timeout) => {
                    if newer_cluster_snapshot_timeout.elapsed() > NEWER_SNAPSHOT_THRESHOLD {
                        return Err(GetRpcNodeError::NoNewerSnapshots);
                    }
                }
            }
            retry_reason = Some("No snapshots available".to_owned());
            continue;
        } else {
            let rpc_peers = peer_snapshot_hashes
                .iter()
                .map(|peer_snapshot_hash| peer_snapshot_hash.rpc_contact_info.id)
                .collect::<Vec<_>>();
            let final_snapshot_hash = peer_snapshot_hashes[0].snapshot_hash;
            info!(
                "Highest available snapshot slot is {}, available from {} node{}: {:?}",
                final_snapshot_hash
                    .incr
                    .map(|(slot, _hash)| slot)
                    .unwrap_or(final_snapshot_hash.full.0),
                rpc_peers.len(),
                if rpc_peers.len() > 1 { "s" } else { "" },
                rpc_peers,
            );
            let rpc_node_results = peer_snapshot_hashes
                .iter()
                .map(|peer_snapshot_hash| GetRpcNodeResult {
                    rpc_contact_info: peer_snapshot_hash.rpc_contact_info.clone(),
                    snapshot_hash: Some(peer_snapshot_hash.snapshot_hash),
                })
                .take(MAX_RPC_CONNECTIONS_EVALUATED_PER_ITERATION)
                .collect();
            return Ok(rpc_node_results);
        }
    }
}

/// Get the Slot and Hash of the local snapshot with the highest slot.  Can be either a full
/// snapshot or an incremental snapshot.
fn get_highest_local_snapshot_hash(
    full_snapshot_archives_dir: impl AsRef<Path>,
    incremental_snapshot_archives_dir: impl AsRef<Path>,
    incremental_snapshot_fetch: bool,
) -> Option<(Slot, Hash)> {
    snapshot_utils::get_highest_full_snapshot_archive_info(full_snapshot_archives_dir)
        .and_then(|full_snapshot_info| {
            if incremental_snapshot_fetch {
                snapshot_utils::get_highest_incremental_snapshot_archive_info(
                    incremental_snapshot_archives_dir,
                    full_snapshot_info.slot(),
                )
                .map(|incremental_snapshot_info| {
                    (
                        incremental_snapshot_info.slot(),
                        *incremental_snapshot_info.hash(),
                    )
                })
            } else {
                None
            }
            .or_else(|| Some((full_snapshot_info.slot(), *full_snapshot_info.hash())))
        })
        .map(|(slot, snapshot_hash)| (slot, snapshot_hash.0))
}

/// Get peer snapshot hashes
///
/// The result is a vector of peers with snapshot hashes that:
/// 1. match a snapshot hash from the known validators
/// 2. have the highest incremental snapshot slot
/// 3. have the highest full snapshot slot of (2)
fn get_peer_snapshot_hashes(
    cluster_info: &ClusterInfo,
    rpc_peers: &[ContactInfo],
    known_validators: Option<&HashSet<Pubkey>>,
    known_validators_to_wait_for: KnownValidatorsToWaitFor,
    incremental_snapshot_fetch: bool,
) -> Vec<PeerSnapshotHash> {
    let mut peer_snapshot_hashes =
        get_eligible_peer_snapshot_hashes(cluster_info, rpc_peers, incremental_snapshot_fetch);
    if let Some(known_validators) = known_validators {
        let known_snapshot_hashes = get_snapshot_hashes_from_known_validators(
            cluster_info,
            known_validators,
            known_validators_to_wait_for,
            incremental_snapshot_fetch,
        );
        retain_peer_snapshot_hashes_that_match_known_snapshot_hashes(
            &known_snapshot_hashes,
            &mut peer_snapshot_hashes,
        );
    }
    retain_peer_snapshot_hashes_with_highest_incremental_snapshot_slot(&mut peer_snapshot_hashes);
    retain_peer_snapshot_hashes_with_highest_full_snapshot_slot(&mut peer_snapshot_hashes);

    peer_snapshot_hashes
}

/// Map full snapshot hashes to a set of incremental snapshot hashes.  Each full snapshot hash
/// is treated as the base for its set of incremental snapshot hashes.
type KnownSnapshotHashes = HashMap<(Slot, Hash), HashSet<(Slot, Hash)>>;

/// Get the snapshot hashes from known validators.
///
/// The snapshot hashes are put into a map from full snapshot hash to a set of incremental
/// snapshot hashes.  This map will be used as the "known snapshot hashes"; when peers are
/// queried for their individual snapshot hashes, their results will be checked against this
/// map to verify correctness.
///
/// NOTE: Only a single snashot hash is allowed per slot.  If somehow two known validators have
/// a snapshot hash with the same slot and _different_ hashes, the second will be skipped.
/// This applies to both full and incremental snapshot hashes.
fn get_snapshot_hashes_from_known_validators(
    cluster_info: &ClusterInfo,
    known_validators: &HashSet<Pubkey>,
    known_validators_to_wait_for: KnownValidatorsToWaitFor,
    incremental_snapshot_fetch: bool,
) -> KnownSnapshotHashes {
    // Get the full snapshot hashes for a node from CRDS
    let get_full_snapshot_hashes_for_node = |node| {
        let mut full_snapshot_hashes = Vec::new();
        cluster_info.get_legacy_snapshot_hash_for_node(node, |snapshot_hashes| {
            full_snapshot_hashes = snapshot_hashes.clone();
        });
        full_snapshot_hashes
    };

    // Get the incremental snapshot hashes for a node from CRDS
    let get_incremental_snapshot_hashes_for_node = |node| {
        cluster_info
            .get_snapshot_hashes_for_node(node)
            .map(|hashes| (hashes.full, hashes.incremental))
    };

    if !do_known_validators_have_all_snapshot_hashes(
        known_validators,
        known_validators_to_wait_for,
        get_full_snapshot_hashes_for_node,
        get_incremental_snapshot_hashes_for_node,
        incremental_snapshot_fetch,
    ) {
        debug!(
            "Snapshot hashes have not been discovered from known validators. \
            This likely means the gossip tables are not fully populated. \
            We will sleep and retry..."
        );
        return KnownSnapshotHashes::default();
    }

    build_known_snapshot_hashes(
        known_validators,
        get_full_snapshot_hashes_for_node,
        get_incremental_snapshot_hashes_for_node,
        incremental_snapshot_fetch,
    )
}

/// Check if we can discover snapshot hashes for the known validators.
///
/// This is a work-around to ensure the gossip tables are populated enough so that the bootstrap
/// process will download both full and incremental snapshots.  If the incremental snapshot hashes
/// are not yet populated from gossip, then it is possible (and has been seen often) to only
/// discover full snapshots—and ones that are very old (up to 25,000 slots)—but *not* discover any
/// of their associated incremental snapshots.
///
/// This function will return false if we do not yet have snapshot hashes from known validators;
/// and true otherwise.  Either require snapshot hashes from *all* or *any* of the known validators
/// based on the `KnownValidatorsToWaitFor` parameter.
fn do_known_validators_have_all_snapshot_hashes<'a, F1, F2>(
    known_validators: impl IntoIterator<Item = &'a Pubkey>,
    known_validators_to_wait_for: KnownValidatorsToWaitFor,
    get_full_snapshot_hashes_for_node: F1,
    get_incremental_snapshot_hashes_for_node: F2,
    incremental_snapshot_fetch: bool,
) -> bool
where
    F1: Fn(&'a Pubkey) -> Vec<(Slot, Hash)>,
    F2: Fn(&'a Pubkey) -> Option<((Slot, Hash), Vec<(Slot, Hash)>)>,
{
    let node_has_full_snapshot_hashes = |node| !get_full_snapshot_hashes_for_node(node).is_empty();
    let node_has_incremental_snapshot_hashes = |node| {
        get_incremental_snapshot_hashes_for_node(node)
            .map(|(_, hashes)| !hashes.is_empty())
            .unwrap_or(false)
    };

    // Does this node have all the snapshot hashes?
    // If incremental snapshots are disabled, only check for full snapshot hashes; otherwise check
    // for both full and incremental snapshot hashes.
    let node_has_all_snapshot_hashes = |node| {
        node_has_full_snapshot_hashes(node)
            && (!incremental_snapshot_fetch || node_has_incremental_snapshot_hashes(node))
    };

    match known_validators_to_wait_for {
        KnownValidatorsToWaitFor::All => known_validators
            .into_iter()
            .all(node_has_all_snapshot_hashes),
        KnownValidatorsToWaitFor::Any => known_validators
            .into_iter()
            .any(node_has_all_snapshot_hashes),
    }
}

/// When waiting for snapshot hashes from the known validators, should we wait for *all* or *any*
/// of them?
#[derive(Debug, Copy, Clone, Eq, PartialEq)]
enum KnownValidatorsToWaitFor {
    All,
    Any,
}

/// Build the known snapshot hashes from a set of nodes.
///
/// The `get_full_snapshot_hashes_for_node` and `get_incremental_snapshot_hashes_for_node`
/// parameters are Fns that map a pubkey to its respective full and incremental snapshot
/// hashes.  These parameters exist to provide a way to test the inner algorithm without
/// needing runtime information such as the ClusterInfo or ValidatorConfig.
fn build_known_snapshot_hashes<'a, F1, F2>(
    nodes: impl IntoIterator<Item = &'a Pubkey>,
    get_full_snapshot_hashes_for_node: F1,
    get_incremental_snapshot_hashes_for_node: F2,
    incremental_snapshot_fetch: bool,
) -> KnownSnapshotHashes
where
    F1: Fn(&'a Pubkey) -> Vec<(Slot, Hash)>,
    F2: Fn(&'a Pubkey) -> Option<((Slot, Hash), Vec<(Slot, Hash)>)>,
{
    let mut known_snapshot_hashes = KnownSnapshotHashes::new();

    /// Check to see if there exists another snapshot hash in the haystack with the *same* slot
    /// but *different* hash as the needle.
    fn is_any_same_slot_and_different_hash<'a>(
        needle: &(Slot, Hash),
        haystack: impl IntoIterator<Item = &'a (Slot, Hash)>,
    ) -> bool {
        haystack
            .into_iter()
            .any(|hay| needle.0 == hay.0 && needle.1 != hay.1)
    }

    'to_next_node: for node in nodes {
        // First get the full snapshot hashes for each node and add them as the keys in the
        // known snapshot hashes map.
        let full_snapshot_hashes = get_full_snapshot_hashes_for_node(node);
        '_to_next_full_snapshot: for full_snapshot_hash in &full_snapshot_hashes {
            // Do not add this snapshot hash if there's already a full snapshot hash with the
            // same slot but with a _different_ hash.
            // NOTE: Nodes should not produce snapshots at the same slot with _different_
            // hashes.  So if it happens, keep the first and ignore the rest.
            if is_any_same_slot_and_different_hash(full_snapshot_hash, known_snapshot_hashes.keys())
            {
                warn!(
                    "Ignoring all snapshot hashes from node {} since we've seen a different full snapshot hash with this slot.\nfull snapshot hash: {:?}",
                    node,
                    full_snapshot_hash,
                );
                debug!(
                    "known full snapshot hashes: {:#?}",
                    known_snapshot_hashes.keys(),
                );
                continue 'to_next_node;
            }

            // Insert a new full snapshot hash into the known snapshot hashes IFF an entry
            // doesn't already exist.  This is to ensure we don't overwrite existing
            // incremental snapshot hashes that may be present for this full snapshot hash.
            let _ = known_snapshot_hashes
                .entry(*full_snapshot_hash)
                .or_default();
        }

        if incremental_snapshot_fetch {
            // Then get the incremental snapshot hashes for each node and add them as the values in the
            // known snapshot hashes map.
            if let Some((base_snapshot_hash, incremental_snapshot_hashes)) =
                get_incremental_snapshot_hashes_for_node(node)
            {
                // Incremental snapshots must be based off a valid full snapshot.  Ensure the node
                // has a full snapshot hash that matches its base snapshot hash.
                if !full_snapshot_hashes.contains(&base_snapshot_hash) {
                    warn!(
                        "Ignoring all incremental snapshot hashes from node {} since its base snapshot hash does not match any of its full snapshot hashes.\nbase snapshot hash: {:?}\nfull snapshot hashes: {:?}",
                        node,
                        base_snapshot_hash,
                        full_snapshot_hashes
                    );
                    continue 'to_next_node;
                }

                if let Some(known_incremental_snapshot_hashes) =
                    known_snapshot_hashes.get_mut(&base_snapshot_hash)
                {
                    'to_next_incremental_snapshot: for incremental_snapshot_hash in
                        &incremental_snapshot_hashes
                    {
                        // Do not add this snapshot hash if there's already an incremental snapshot
                        // hash with the same slot, but with a _different_ hash.
                        // NOTE: Nodes should not produce snapshots at the same slot with _different_
                        // hashes.  So if it happens, keep the first and ignore the rest.
                        if is_any_same_slot_and_different_hash(
                            incremental_snapshot_hash,
                            known_incremental_snapshot_hashes.iter(),
                        ) {
                            warn!(
                                "Ignoring incremental snapshot hash from node {} since we've seen a different incremental snapshot hash with this slot.\nbase snapshot hash: {:?}\nincremental snapshot hash: {:?}",
                                node,
                                base_snapshot_hash,
                                incremental_snapshot_hash,
                            );
                            debug!(
                                "known incremental snapshot hashes at this slot: {:#?}",
                                known_incremental_snapshot_hashes.iter(),
                            );
                            continue 'to_next_incremental_snapshot;
                        }

                        known_incremental_snapshot_hashes.insert(*incremental_snapshot_hash);
                    }
                } else {
                    // Since incremental snapshots *must* have a valid base (i.e. full)
                    // snapshot, if .get() returned None, then that can only happen if there
                    // already is a full snapshot hash in the known snapshot hashes with the
                    // same slot but _different_ a hash.  Assert that below.  If the assert
                    // ever fails, there is a programmer bug.
                    assert!(
                        is_any_same_slot_and_different_hash(&base_snapshot_hash, known_snapshot_hashes.keys()),
                        "There must exist a full snapshot hash already in the known snapshot hashes with the same slot but a different hash!",
                    );
                    debug!(
                        "Ignoring incremental snapshot hashes from node {} since we've seen a different base snapshot hash with this slot.\nbase snapshot hash: {:?}\nknown full snapshot hashes: {:?}",
                        node,
                        base_snapshot_hash,
                        known_snapshot_hashes.keys(),
                    );
                    continue 'to_next_node;
                }
            }
        }
    }

    trace!("known snapshot hashes: {:?}", &known_snapshot_hashes);
    known_snapshot_hashes
}

/// Get snapshot hashes from all the eligible peers.  This fn will get only one
/// snapshot hash per peer (the one with the highest slot).  This may be just a full snapshot
/// hash, or a combo full (i.e. base) snapshot hash and incremental snapshot hash.
fn get_eligible_peer_snapshot_hashes(
    cluster_info: &ClusterInfo,
    rpc_peers: &[ContactInfo],
    incremental_snapshot_fetch: bool,
) -> Vec<PeerSnapshotHash> {
    let mut peer_snapshot_hashes = Vec::new();
    for rpc_peer in rpc_peers {
        // Get this peer's highest (full) snapshot hash.  We need to get these snapshot hashes
        // (instead of just the IncrementalSnapshotHashes) in case the peer is either (1) not
        // taking incremental snapshots, or (2) if the last snapshot taken was a full snapshot,
        // which would get pushed to CRDS here (i.e. `crds_value::SnapshotHashes`) first.
        let highest_snapshot_hash =
            get_highest_full_snapshot_hash_for_peer(cluster_info, &rpc_peer.id).max(
                if incremental_snapshot_fetch {
                    get_highest_incremental_snapshot_hash_for_peer(cluster_info, &rpc_peer.id)
                } else {
                    None
                },
            );

        if let Some(snapshot_hash) = highest_snapshot_hash {
            peer_snapshot_hashes.push(PeerSnapshotHash {
                rpc_contact_info: rpc_peer.clone(),
                snapshot_hash,
            });
        };
    }

    trace!("peer snapshot hashes: {:?}", &peer_snapshot_hashes);
    peer_snapshot_hashes
}

/// Retain the peer snapshot hashes that match a snapshot hash from the known snapshot hashes
fn retain_peer_snapshot_hashes_that_match_known_snapshot_hashes(
    known_snapshot_hashes: &KnownSnapshotHashes,
    peer_snapshot_hashes: &mut Vec<PeerSnapshotHash>,
) {
    peer_snapshot_hashes.retain(|peer_snapshot_hash| {
        known_snapshot_hashes
            .get(&peer_snapshot_hash.snapshot_hash.full)
            .map(|known_incremental_hashes| {
                if peer_snapshot_hash.snapshot_hash.incr.is_none() {
                    // If the peer's full snapshot hashes match, but doesn't have any
                    // incremental snapshots, that's fine; keep 'em!
                    true
                } else {
                    known_incremental_hashes
                        .contains(peer_snapshot_hash.snapshot_hash.incr.as_ref().unwrap())
                }
            })
            .unwrap_or(false)
    });

    trace!(
        "retain peer snapshot hashes that match known snapshot hashes: {:?}",
        &peer_snapshot_hashes
    );
}

/// Retain the peer snapshot hashes with the highest full snapshot slot
fn retain_peer_snapshot_hashes_with_highest_full_snapshot_slot(
    peer_snapshot_hashes: &mut Vec<PeerSnapshotHash>,
) {
    // retain the hashes with the highest full snapshot slot
    // do a two-pass algorithm
    // 1. find the full snapshot hash with the highest full snapshot slot
    // 2. retain elems with that full snapshot hash
    let mut highest_full_snapshot_hash = (Slot::MIN, Hash::default());
    peer_snapshot_hashes.iter().for_each(|peer_snapshot_hash| {
        if peer_snapshot_hash.snapshot_hash.full.0 > highest_full_snapshot_hash.0 {
            highest_full_snapshot_hash = peer_snapshot_hash.snapshot_hash.full;
        }
    });

    peer_snapshot_hashes.retain(|peer_snapshot_hash| {
        peer_snapshot_hash.snapshot_hash.full == highest_full_snapshot_hash
    });

    trace!(
        "retain peer snapshot hashes with highest full snapshot slot: {:?}",
        &peer_snapshot_hashes
    );
}

/// Retain the peer snapshot hashes with the highest incremental snapshot slot
fn retain_peer_snapshot_hashes_with_highest_incremental_snapshot_slot(
    peer_snapshot_hashes: &mut Vec<PeerSnapshotHash>,
) {
    let mut highest_incremental_snapshot_hash: Option<(Slot, Hash)> = None;
    peer_snapshot_hashes.iter().for_each(|peer_snapshot_hash| {
        if let Some(incremental_snapshot_hash) = peer_snapshot_hash.snapshot_hash.incr.as_ref() {
            if highest_incremental_snapshot_hash.is_none()
                || incremental_snapshot_hash.0 > highest_incremental_snapshot_hash.unwrap().0
            {
                highest_incremental_snapshot_hash = Some(*incremental_snapshot_hash);
            }
        };
    });

    peer_snapshot_hashes.retain(|peer_snapshot_hash| {
        peer_snapshot_hash.snapshot_hash.incr == highest_incremental_snapshot_hash
    });

    trace!(
        "retain peer snapshot hashes with highest incremental snapshot slot: {:?}",
        &peer_snapshot_hashes
    );
}

/// Check to see if we can use our local snapshots, otherwise download newer ones.
#[allow(clippy::too_many_arguments)]
fn download_snapshots(
    full_snapshot_archives_dir: &Path,
    incremental_snapshot_archives_dir: &Path,
    validator_config: &ValidatorConfig,
    bootstrap_config: &RpcBootstrapConfig,
    use_progress_bar: bool,
    maximum_local_snapshot_age: Slot,
    start_progress: &Arc<RwLock<ValidatorStartProgress>>,
    minimal_snapshot_download_speed: f32,
    maximum_snapshot_download_abort: u64,
    download_abort_count: &mut u64,
    snapshot_hash: Option<SnapshotHash>,
    rpc_contact_info: &ContactInfo,
) -> Result<(), String> {
    if snapshot_hash.is_none() {
        return Ok(());
    }
    let SnapshotHash {
        full: full_snapshot_hash,
        incr: incremental_snapshot_hash,
    } = snapshot_hash.unwrap();

    // If the local snapshots are new enough, then use 'em; no need to download new snapshots
    if should_use_local_snapshot(
        full_snapshot_archives_dir,
        incremental_snapshot_archives_dir,
        maximum_local_snapshot_age,
        full_snapshot_hash,
        incremental_snapshot_hash,
        bootstrap_config.incremental_snapshot_fetch,
    ) {
        return Ok(());
    }

    // Check and see if we've already got the full snapshot; if not, download it
    if snapshot_utils::get_full_snapshot_archives(full_snapshot_archives_dir)
        .into_iter()
        .any(|snapshot_archive| {
            snapshot_archive.slot() == full_snapshot_hash.0
                && snapshot_archive.hash().0 == full_snapshot_hash.1
        })
    {
        info!(
            "Full snapshot archive already exists locally. Skipping download. slot: {}, hash: {}",
            full_snapshot_hash.0, full_snapshot_hash.1
        );
    } else {
        download_snapshot(
            full_snapshot_archives_dir,
            incremental_snapshot_archives_dir,
            validator_config,
            bootstrap_config,
            use_progress_bar,
            start_progress,
            minimal_snapshot_download_speed,
            maximum_snapshot_download_abort,
            download_abort_count,
            rpc_contact_info,
            full_snapshot_hash,
            SnapshotType::FullSnapshot,
        )?;
    }

    // Check and see if we've already got the incremental snapshot; if not, download it
    if let Some(incremental_snapshot_hash) = incremental_snapshot_hash {
        if snapshot_utils::get_incremental_snapshot_archives(incremental_snapshot_archives_dir)
            .into_iter()
            .any(|snapshot_archive| {
                snapshot_archive.slot() == incremental_snapshot_hash.0
                    && snapshot_archive.hash().0 == incremental_snapshot_hash.1
                    && snapshot_archive.base_slot() == full_snapshot_hash.0
            })
        {
            info!(
                "Incremental snapshot archive already exists locally. Skipping download. slot: {}, hash: {}",
                incremental_snapshot_hash.0, incremental_snapshot_hash.1
            );
        } else {
            download_snapshot(
                full_snapshot_archives_dir,
                incremental_snapshot_archives_dir,
                validator_config,
                bootstrap_config,
                use_progress_bar,
                start_progress,
                minimal_snapshot_download_speed,
                maximum_snapshot_download_abort,
                download_abort_count,
                rpc_contact_info,
                incremental_snapshot_hash,
                SnapshotType::IncrementalSnapshot(full_snapshot_hash.0),
            )?;
        }
    }

    Ok(())
}

/// Download a snapshot
#[allow(clippy::too_many_arguments)]
fn download_snapshot(
    full_snapshot_archives_dir: &Path,
    incremental_snapshot_archives_dir: &Path,
    validator_config: &ValidatorConfig,
    bootstrap_config: &RpcBootstrapConfig,
    use_progress_bar: bool,
    start_progress: &Arc<RwLock<ValidatorStartProgress>>,
    minimal_snapshot_download_speed: f32,
    maximum_snapshot_download_abort: u64,
    download_abort_count: &mut u64,
    rpc_contact_info: &ContactInfo,
    desired_snapshot_hash: (Slot, Hash),
    snapshot_type: SnapshotType,
) -> Result<(), String> {
    let maximum_full_snapshot_archives_to_retain = validator_config
        .snapshot_config
        .maximum_full_snapshot_archives_to_retain;
    let maximum_incremental_snapshot_archives_to_retain = validator_config
        .snapshot_config
        .maximum_incremental_snapshot_archives_to_retain;

    *start_progress.write().unwrap() = ValidatorStartProgress::DownloadingSnapshot {
        slot: desired_snapshot_hash.0,
        rpc_addr: rpc_contact_info.rpc,
    };
    let desired_snapshot_hash = (
        desired_snapshot_hash.0,
        solana_runtime::snapshot_hash::SnapshotHash(desired_snapshot_hash.1),
    );
    download_snapshot_archive(
        &rpc_contact_info.rpc,
        full_snapshot_archives_dir,
        incremental_snapshot_archives_dir,
        desired_snapshot_hash,
        snapshot_type,
        maximum_full_snapshot_archives_to_retain,
        maximum_incremental_snapshot_archives_to_retain,
        use_progress_bar,
        &mut Some(Box::new(|download_progress: &DownloadProgressRecord| {
            debug!("Download progress: {:?}", download_progress);
            if download_progress.last_throughput < minimal_snapshot_download_speed
                && download_progress.notification_count <= 1
                && download_progress.percentage_done <= 2_f32
                && download_progress.estimated_remaining_time > 60_f32
                && *download_abort_count < maximum_snapshot_download_abort
            {
                if let Some(ref known_validators) = validator_config.known_validators {
                    if known_validators.contains(&rpc_contact_info.id)
                        && known_validators.len() == 1
                        && bootstrap_config.only_known_rpc
                    {
                        warn!(
                            "The snapshot download is too slow, throughput: {} < min speed {} \
                            bytes/sec, but will NOT abort and try a different node as it is the \
                            only known validator and the --only-known-rpc flag is set. \
                            Abort count: {}, Progress detail: {:?}",
                            download_progress.last_throughput,
                            minimal_snapshot_download_speed,
                            download_abort_count,
                            download_progress,
                        );
                        return true; // Do not abort download from the one-and-only known validator
                    }
                }
                warn!(
                    "The snapshot download is too slow, throughput: {} < min speed {} \
                    bytes/sec, will abort and try a different node. \
                    Abort count: {}, Progress detail: {:?}",
                    download_progress.last_throughput,
                    minimal_snapshot_download_speed,
                    download_abort_count,
                    download_progress,
                );
                *download_abort_count += 1;
                false
            } else {
                true
            }
        })),
    )
}

/// Check to see if bootstrap should load from its local snapshots or not.  If not, then snapshots
/// will be downloaded.
fn should_use_local_snapshot(
    full_snapshot_archives_dir: &Path,
    incremental_snapshot_archives_dir: &Path,
    maximum_local_snapshot_age: Slot,
    full_snapshot_hash: (Slot, Hash),
    incremental_snapshot_hash: Option<(Slot, Hash)>,
    incremental_snapshot_fetch: bool,
) -> bool {
    let cluster_snapshot_slot = incremental_snapshot_hash
        .map(|(slot, _)| slot)
        .unwrap_or(full_snapshot_hash.0);

    match get_highest_local_snapshot_hash(
        full_snapshot_archives_dir,
        incremental_snapshot_archives_dir,
        incremental_snapshot_fetch,
    ) {
        None => {
            info!(
                "Downloading a snapshot for slot {} since there is not a local snapshot.",
                cluster_snapshot_slot,
            );
            false
        }
        Some((local_snapshot_slot, _)) => {
            if local_snapshot_slot
                >= cluster_snapshot_slot.saturating_sub(maximum_local_snapshot_age)
            {
                info!(
                    "Reusing local snapshot at slot {} instead of downloading a snapshot for slot {}.",
                    local_snapshot_slot,
                    cluster_snapshot_slot,
                );
                true
            } else {
                info!(
                    "Local snapshot from slot {} is too old. Downloading a newer snapshot for slot {}.",
                    local_snapshot_slot,
                    cluster_snapshot_slot,
                );
                false
            }
        }
    }
}

/// Get the highest full snapshot hash for a peer from CRDS
fn get_highest_full_snapshot_hash_for_peer(
    cluster_info: &ClusterInfo,
    peer: &Pubkey,
) -> Option<SnapshotHash> {
    let mut full_snapshot_hashes = Vec::new();
    cluster_info.get_legacy_snapshot_hash_for_node(peer, |snapshot_hashes| {
        full_snapshot_hashes = snapshot_hashes.clone()
    });
    full_snapshot_hashes
        .into_iter()
        .max()
        .map(|full_snapshot_hash| SnapshotHash {
            full: full_snapshot_hash,
            incr: None,
        })
}

/// Get the highest incremental snapshot hash for a peer from CRDS
fn get_highest_incremental_snapshot_hash_for_peer(
    cluster_info: &ClusterInfo,
    peer: &Pubkey,
) -> Option<SnapshotHash> {
    cluster_info.get_snapshot_hashes_for_node(peer).map(
        |crds_value::SnapshotHashes {
             full, incremental, ..
         }| {
            let highest_incremental_snapshot_hash = incremental.into_iter().max();
            SnapshotHash {
                full,
                incr: highest_incremental_snapshot_hash,
            }
        },
    )
}

#[cfg(test)]
mod tests {
    use super::*;

    impl PeerSnapshotHash {
        fn new(
            rpc_contact_info: ContactInfo,
            full_snapshot_hash: (Slot, Hash),
            incremental_snapshot_hash: Option<(Slot, Hash)>,
        ) -> Self {
            Self {
                rpc_contact_info,
                snapshot_hash: SnapshotHash {
                    full: full_snapshot_hash,
                    incr: incremental_snapshot_hash,
                },
            }
        }
    }

    fn default_contact_info_for_tests() -> ContactInfo {
        let sock_addr = SocketAddr::from(([1, 1, 1, 1], 11_111));
        ContactInfo {
            id: Pubkey::default(),
            gossip: sock_addr,
            tvu: sock_addr,
            tvu_forwards: sock_addr,
            repair: sock_addr,
            tpu: sock_addr,
            tpu_forwards: sock_addr,
            tpu_vote: sock_addr,
            rpc: sock_addr,
            rpc_pubsub: sock_addr,
            serve_repair: sock_addr,
            wallclock: 123456789,
            shred_version: 1,
        }
    }

    #[test]
    fn test_build_known_snapshot_hashes() {
        let full_snapshot_hashes1 = vec![
            (100_000, Hash::new_unique()),
            (200_000, Hash::new_unique()),
            (300_000, Hash::new_unique()),
            (400_000, Hash::new_unique()),
        ];
        let full_snapshot_hashes2 = vec![
            (100_000, Hash::new_unique()),
            (200_000, Hash::new_unique()),
            (300_000, Hash::new_unique()),
            (400_000, Hash::new_unique()),
        ];

        let base_snapshot_hash1 = full_snapshot_hashes1.last().unwrap();
        let base_snapshot_hash2 = full_snapshot_hashes2.last().unwrap();

        let incremental_snapshot_hashes1 = vec![
            (400_500, Hash::new_unique()),
            (400_600, Hash::new_unique()),
            (400_700, Hash::new_unique()),
            (400_800, Hash::new_unique()),
        ];
        let incremental_snapshot_hashes2 = vec![
            (400_500, Hash::new_unique()),
            (400_600, Hash::new_unique()),
            (400_700, Hash::new_unique()),
            (400_800, Hash::new_unique()),
        ];

        #[allow(clippy::type_complexity)]
        let mut oracle: HashMap<
            Pubkey,
            (Vec<(Slot, Hash)>, Option<((Slot, Hash), Vec<(Slot, Hash)>)>),
        > = HashMap::new();

        // no snapshots at all
        oracle.insert(Pubkey::new_unique(), (vec![], None));

        // just full snapshots
        oracle.insert(Pubkey::new_unique(), (full_snapshot_hashes1.clone(), None));

        // just full snapshots, with different hashes
        oracle.insert(Pubkey::new_unique(), (full_snapshot_hashes2.clone(), None));

        // full and incremental snapshots
        oracle.insert(
            Pubkey::new_unique(),
            (
                full_snapshot_hashes1.clone(),
                Some((*base_snapshot_hash1, incremental_snapshot_hashes1.clone())),
            ),
        );

        // full and incremental snapshots, but base hash is wrong
        oracle.insert(
            Pubkey::new_unique(),
            (
                full_snapshot_hashes1.clone(),
                Some((*base_snapshot_hash2, incremental_snapshot_hashes1.clone())),
            ),
        );

        // full and incremental snapshots, with different incremental hashes
        oracle.insert(
            Pubkey::new_unique(),
            (
                full_snapshot_hashes1.clone(),
                Some((*base_snapshot_hash1, incremental_snapshot_hashes2.clone())),
            ),
        );

        // full and incremental snapshots, with different hashes
        oracle.insert(
            Pubkey::new_unique(),
            (
                full_snapshot_hashes2.clone(),
                Some((*base_snapshot_hash2, incremental_snapshot_hashes2.clone())),
            ),
        );

        // full and incremental snapshots, but base hash is wrong
        oracle.insert(
            Pubkey::new_unique(),
            (
                full_snapshot_hashes2.clone(),
                Some((*base_snapshot_hash1, incremental_snapshot_hashes2.clone())),
            ),
        );

        // full and incremental snapshots, with different incremental hashes
        oracle.insert(
            Pubkey::new_unique(),
            (
                full_snapshot_hashes2.clone(),
                Some((*base_snapshot_hash2, incremental_snapshot_hashes1.clone())),
            ),
        );

        // handle duplicates as well
        oracle.insert(
            Pubkey::new_unique(),
            (
                full_snapshot_hashes1.clone(),
                Some((*base_snapshot_hash1, incremental_snapshot_hashes1.clone())),
            ),
        );

        // handle duplicates as well, with different incremental hashes
        oracle.insert(
            Pubkey::new_unique(),
            (
                full_snapshot_hashes1.clone(),
                Some((*base_snapshot_hash1, incremental_snapshot_hashes2.clone())),
            ),
        );

        // handle duplicates, with different hashes
        oracle.insert(
            Pubkey::new_unique(),
            (
                full_snapshot_hashes2.clone(),
                Some((*base_snapshot_hash2, incremental_snapshot_hashes2.clone())),
            ),
        );

        // handle duplicates, with different incremental hashes
        oracle.insert(
            Pubkey::new_unique(),
            (
                full_snapshot_hashes2.clone(),
                Some((*base_snapshot_hash2, incremental_snapshot_hashes1.clone())),
            ),
        );

        let node_to_full_snapshot_hashes = |node| oracle.get(node).unwrap().clone().0;
        let node_to_incremental_snapshot_hashes = |node| oracle.get(node).unwrap().clone().1;

        // With incremental snapshots
        {
            let known_snapshot_hashes = build_known_snapshot_hashes(
                oracle.keys(),
                node_to_full_snapshot_hashes,
                node_to_incremental_snapshot_hashes,
                true,
            );

            let mut known_full_snapshot_hashes: Vec<_> =
                known_snapshot_hashes.keys().copied().collect();
            known_full_snapshot_hashes.sort_unstable();

            let known_base_snapshot_hash = known_full_snapshot_hashes.last().unwrap();

            let mut known_incremental_snapshot_hashes: Vec<_> = known_snapshot_hashes
                .get(known_base_snapshot_hash)
                .unwrap()
                .iter()
                .copied()
                .collect();
            known_incremental_snapshot_hashes.sort_unstable();

            // The resulting `known_snapshot_hashes` can be different from run-to-run due to how
            // `oracle.keys()` returns nodes during iteration.  Because of that, we cannot just assert
            // the full and incremental snapshot hashes are `full_snapshot_hashes1` and
            // `incremental_snapshot_hashes2`.  Instead, we assert that the full and incremental
            // snapshot hashes are exactly one or the other, since it depends on which nodes are seen
            // "first" when building the known snapshot hashes.

            assert!(
                known_full_snapshot_hashes == full_snapshot_hashes1
                    || known_full_snapshot_hashes == full_snapshot_hashes2
            );

            if known_full_snapshot_hashes == full_snapshot_hashes1 {
                assert_eq!(known_base_snapshot_hash, base_snapshot_hash1);
            } else {
                assert_eq!(known_base_snapshot_hash, base_snapshot_hash2);
            }

            assert!(
                known_incremental_snapshot_hashes == incremental_snapshot_hashes1
                    || known_incremental_snapshot_hashes == incremental_snapshot_hashes2
            );
        }

        // Without incremental snapshots
        {
            let known_snapshot_hashes = build_known_snapshot_hashes(
                oracle.keys(),
                node_to_full_snapshot_hashes,
                node_to_incremental_snapshot_hashes,
                false,
            );

            let mut known_full_snapshot_hashes: Vec<_> =
                known_snapshot_hashes.keys().copied().collect();
            known_full_snapshot_hashes.sort_unstable();

            let known_base_snapshot_hash = known_full_snapshot_hashes.last().unwrap();

            let known_incremental_snapshot_hashes =
                known_snapshot_hashes.get(known_base_snapshot_hash).unwrap();

            // The resulting `known_snapshot_hashes` can be different from run-to-run due to how
            // `oracle.keys()` returns nodes during iteration.  Because of that, we cannot just
            // assert the full snapshot hashes are `full_snapshot_hashes1`.  Instead, we assert
            // that the full snapshot hashes are exactly one or the other, since it depends on
            // which nodes are seen "first" when building the known snapshot hashes.

            assert!(
                known_full_snapshot_hashes == full_snapshot_hashes1
                    || known_full_snapshot_hashes == full_snapshot_hashes2
            );

            assert!(known_incremental_snapshot_hashes.is_empty());
        }
    }

    #[test]
    fn test_retain_peer_snapshot_hashes_that_match_known_snapshot_hashes() {
        let known_snapshot_hashes: KnownSnapshotHashes = [
            (
                (200_000, Hash::new_unique()),
                [
                    (200_200, Hash::new_unique()),
                    (200_400, Hash::new_unique()),
                    (200_600, Hash::new_unique()),
                    (200_800, Hash::new_unique()),
                ]
                .iter()
                .cloned()
                .collect(),
            ),
            (
                (300_000, Hash::new_unique()),
                [
                    (300_200, Hash::new_unique()),
                    (300_400, Hash::new_unique()),
                    (300_600, Hash::new_unique()),
                ]
                .iter()
                .cloned()
                .collect(),
            ),
        ]
        .iter()
        .cloned()
        .collect();

        let known_snapshot_hash = known_snapshot_hashes.iter().next().unwrap();
        let known_full_snapshot_hash = known_snapshot_hash.0;
        let known_incremental_snapshot_hash = known_snapshot_hash.1.iter().next().unwrap();

        let contact_info = default_contact_info_for_tests();
        let peer_snapshot_hashes = vec![
            // bad full snapshot hash, no incremental snapshot hash
            PeerSnapshotHash::new(contact_info.clone(), (111_000, Hash::default()), None),
            // bad everything
            PeerSnapshotHash::new(
                contact_info.clone(),
                (111_000, Hash::default()),
                Some((111_111, Hash::default())),
            ),
            // good full snapshot hash, no incremental snapshot hash
            PeerSnapshotHash::new(contact_info.clone(), *known_full_snapshot_hash, None),
            // bad full snapshot hash, good (not possible) incremental snapshot hash
            PeerSnapshotHash::new(
                contact_info.clone(),
                (111_000, Hash::default()),
                Some(*known_incremental_snapshot_hash),
            ),
            // good full snapshot hash, bad incremental snapshot hash
            PeerSnapshotHash::new(
                contact_info.clone(),
                *known_full_snapshot_hash,
                Some((111_111, Hash::default())),
            ),
            // good everything
            PeerSnapshotHash::new(
                contact_info.clone(),
                *known_full_snapshot_hash,
                Some(*known_incremental_snapshot_hash),
            ),
        ];

        let expected = vec![
            PeerSnapshotHash::new(contact_info.clone(), *known_full_snapshot_hash, None),
            PeerSnapshotHash::new(
                contact_info,
                *known_full_snapshot_hash,
                Some(*known_incremental_snapshot_hash),
            ),
        ];
        let mut actual = peer_snapshot_hashes;
        retain_peer_snapshot_hashes_that_match_known_snapshot_hashes(
            &known_snapshot_hashes,
            &mut actual,
        );
        assert_eq!(expected, actual);
    }

    #[test]
    fn test_retain_peer_snapshot_hashes_with_highest_full_snapshot_slot() {
        let contact_info = default_contact_info_for_tests();
        let peer_snapshot_hashes = vec![
            // old
            PeerSnapshotHash::new(
                contact_info.clone(),
                (100_000, Hash::default()),
                Some((100_100, Hash::default())),
            ),
            PeerSnapshotHash::new(
                contact_info.clone(),
                (100_000, Hash::default()),
                Some((100_200, Hash::default())),
            ),
            PeerSnapshotHash::new(
                contact_info.clone(),
                (100_000, Hash::default()),
                Some((100_300, Hash::default())),
            ),
            // new
            PeerSnapshotHash::new(
                contact_info.clone(),
                (200_000, Hash::default()),
                Some((200_100, Hash::default())),
            ),
            PeerSnapshotHash::new(
                contact_info.clone(),
                (200_000, Hash::default()),
                Some((200_200, Hash::default())),
            ),
            PeerSnapshotHash::new(
                contact_info.clone(),
                (200_000, Hash::default()),
                Some((200_300, Hash::default())),
            ),
        ];

        let expected = vec![
            PeerSnapshotHash::new(
                contact_info.clone(),
                (200_000, Hash::default()),
                Some((200_100, Hash::default())),
            ),
            PeerSnapshotHash::new(
                contact_info.clone(),
                (200_000, Hash::default()),
                Some((200_200, Hash::default())),
            ),
            PeerSnapshotHash::new(
                contact_info,
                (200_000, Hash::default()),
                Some((200_300, Hash::default())),
            ),
        ];
        let mut actual = peer_snapshot_hashes;
        retain_peer_snapshot_hashes_with_highest_full_snapshot_slot(&mut actual);
        assert_eq!(expected, actual);
    }

    #[test]
    fn test_retain_peer_snapshot_hashes_with_highest_incremental_snapshot_slot() {
        let contact_info = default_contact_info_for_tests();
        let peer_snapshot_hashes = vec![
            PeerSnapshotHash::new(contact_info.clone(), (200_000, Hash::default()), None),
            PeerSnapshotHash::new(
                contact_info.clone(),
                (200_000, Hash::default()),
                Some((200_100, Hash::default())),
            ),
            PeerSnapshotHash::new(
                contact_info.clone(),
                (200_000, Hash::default()),
                Some((200_200, Hash::default())),
            ),
            PeerSnapshotHash::new(
                contact_info.clone(),
                (200_000, Hash::default()),
                Some((200_300, Hash::default())),
            ),
            PeerSnapshotHash::new(
                contact_info.clone(),
                (200_000, Hash::default()),
                Some((200_010, Hash::default())),
            ),
            PeerSnapshotHash::new(
                contact_info.clone(),
                (200_000, Hash::default()),
                Some((200_020, Hash::default())),
            ),
            PeerSnapshotHash::new(
                contact_info.clone(),
                (200_000, Hash::default()),
                Some((200_030, Hash::default())),
            ),
        ];

        let expected = vec![PeerSnapshotHash::new(
            contact_info,
            (200_000, Hash::default()),
            Some((200_300, Hash::default())),
        )];
        let mut actual = peer_snapshot_hashes;
        retain_peer_snapshot_hashes_with_highest_incremental_snapshot_slot(&mut actual);
        assert_eq!(expected, actual);
    }
}