shuffle peers before sampling

core/src/repair/repair_service.rs

@@ -23,6 +23,7 @@ use {
     },
     crossbeam_channel::{Receiver as CrossbeamReceiver, Sender as CrossbeamSender},
     lru::LruCache,
+    rand::seq::SliceRandom,
     solana_client::connection_cache::Protocol,
     solana_gossip::cluster_info::ClusterInfo,
     solana_ledger::{
@@ -58,6 +59,12 @@ use {
 const DEFER_REPAIR_THRESHOLD: Duration = Duration::from_millis(200);
 const DEFER_REPAIR_THRESHOLD_TICKS: u64 = DEFER_REPAIR_THRESHOLD.as_millis() as u64 / MS_PER_TICK;
 
+// When requesting repair for a specific shred through the admin RPC, we will
+// request up to NUM_PEERS_TO_SAMPLE_FOR_REPAIRS in the event a specific, valid
+// target node is not provided. This number was chosen to provide reasonable
+// chance of sampling duplicate in the event of cluster partition.
+const NUM_PEERS_TO_SAMPLE_FOR_REPAIRS: usize = 10;
+
 pub type AncestorDuplicateSlotsSender = CrossbeamSender<AncestorDuplicateSlotToRepair>;
 pub type AncestorDuplicateSlotsReceiver = CrossbeamReceiver<AncestorDuplicateSlotToRepair>;
 pub type ConfirmedSlotsSender = CrossbeamSender<Vec<Slot>>;
@@ -683,7 +690,8 @@ impl RepairService {
     }
 
     fn get_repair_peers(cluster_info: Arc<ClusterInfo>, slot: u64) -> Vec<(Pubkey, SocketAddr)> {
-        cluster_info
+        // Collect all nodes likely to have this slot
+        let mut repair_peers: Vec<(Pubkey, SocketAddr)> = cluster_info
             .repair_peers(slot)
             .into_iter()
             .filter_map(|contact_info| {
@@ -693,8 +701,17 @@ impl RepairService {
                     None
                 }
             })
-            .take(10)
-            .collect::<Vec<_>>()
+            .collect();
+
+        // Shuffle the peers to randomize who we request from
+        let mut rng = rand::thread_rng();
+        repair_peers.shuffle(&mut rng);
+
+        // Take a limited number of peers
+        repair_peers
+            .into_iter()
+            .take(NUM_PEERS_TO_SAMPLE_FOR_REPAIRS)
+            .collect()
     }
 
     pub fn request_repair_for_shred_from_peer(
@@ -711,13 +728,16 @@ impl RepairService {
                 cluster_info.lookup_contact_info(&pubkey, |node| node.serve_repair(Protocol::UDP));
             if let Some(peer_repair_addr) = peer_repair_addr {
                 if let Ok(peer_repair_addr) = peer_repair_addr {
+                    trace!("Repair peer {pubkey} has valid repair socket: {peer_repair_addr:?}");
                     repair_peers.push((pubkey, peer_repair_addr));
                 }
             }
         };
         if repair_peers.is_empty() {
-            // No pubkey was provided, or no valid repair socket was found.
-            // sample a set of repair peers instead.
+            debug!(
+                "No pubkey was provided or no valid repair socket was found. \
+                    Sampling a set of repair peers instead."
+            );
             repair_peers = Self::get_repair_peers(cluster_info.clone(), slot);
         }
 

validator/src/main.rs

@@ -794,7 +794,7 @@ pub fn main() {
             return;
         }
         ("repair-shred-from-peer", Some(subcommand_matches)) => {
-            let pubkey = pubkey_of(subcommand_matches, "identity");
+            let pubkey = value_t!(subcommand_matches, "pubkey", Pubkey).ok();
             let slot = value_t_or_exit!(subcommand_matches, "slot", u64);
             let shred_index = value_t_or_exit!(subcommand_matches, "shred", u64);
             let admin_client = admin_rpc_service::connect(&ledger_path);