Add ability to change the validator identity at runtime

core/src/banking_stage.rs

@@ -331,7 +331,6 @@ impl BankingStage {
         // Single thread to generate entries from many banks.
         // This thread talks to poh_service and broadcasts the entries once they have been recorded.
         // Once an entry has been recorded, its blockhash is registered with the bank.
-        let my_pubkey = cluster_info.id();
         let duplicates = Arc::new(Mutex::new((
             LruCache::new(DEFAULT_LRU_SIZE),
             PacketHasher::default(),
@@ -358,7 +357,6 @@ impl BankingStage {
                     .name("solana-banking-stage-tx".to_string())
                     .spawn(move || {
                         Self::process_loop(
-                            my_pubkey,
                             &verified_receiver,
                             &poh_recorder,
                             &cluster_info,
@@ -699,7 +697,6 @@ impl BankingStage {
 
     #[allow(clippy::too_many_arguments)]
     fn process_loop(
-        my_pubkey: Pubkey,
         verified_receiver: &CrossbeamReceiver<Vec<Packets>>,
         poh_recorder: &Arc<Mutex<PohRecorder>>,
         cluster_info: &ClusterInfo,
@@ -718,6 +715,7 @@ impl BankingStage {
         let mut buffered_packets = VecDeque::with_capacity(batch_limit);
         let banking_stage_stats = BankingStageStats::new(id);
         loop {
+            let my_pubkey = cluster_info.id();
             while !buffered_packets.is_empty() {
                 let decision = Self::process_buffered_packets(
                     &my_pubkey,

core/src/broadcast_stage.rs

@@ -173,7 +173,7 @@ pub struct BroadcastStage {
 impl BroadcastStage {
     #[allow(clippy::too_many_arguments)]
     fn run(
-        keypair: &Keypair,
+        cluster_info: Arc<ClusterInfo>,
         blockstore: &Arc<Blockstore>,
         receiver: &Receiver<WorkingBankEntry>,
         socket_sender: &Sender<(TransmitShreds, Option<BroadcastShredBatchInfo>)>,
@@ -182,7 +182,7 @@ impl BroadcastStage {
     ) -> BroadcastStageReturnType {
         loop {
             let res = broadcast_stage_run.run(
-                keypair,
+                &cluster_info.keypair(),
                 blockstore,
                 receiver,
                 socket_sender,
@@ -248,21 +248,23 @@ impl BroadcastStage {
         let bs_run = broadcast_stage_run.clone();
 
         let socket_sender_ = socket_sender.clone();
-        let keypair = cluster_info.keypair.clone();
-        let thread_hdl = Builder::new()
-            .name("solana-broadcaster".to_string())
-            .spawn(move || {
-                let _finalizer = Finalizer::new(exit);
-                Self::run(
-                    &keypair,
-                    &btree,
-                    &receiver,
-                    &socket_sender_,
-                    &blockstore_sender,
-                    bs_run,
-                )
-            })
-            .unwrap();
+        let thread_hdl = {
+            let cluster_info = cluster_info.clone();
+            Builder::new()
+                .name("solana-broadcaster".to_string())
+                .spawn(move || {
+                    let _finalizer = Finalizer::new(exit);
+                    Self::run(
+                        cluster_info,
+                        &btree,
+                        &receiver,
+                        &socket_sender_,
+                        &blockstore_sender,
+                        bs_run,
+                    )
+                })
+                .unwrap()
+        };
         let mut thread_hdls = vec![thread_hdl];
         let socket_receiver = Arc::new(Mutex::new(socket_receiver));
         for sock in socks.into_iter() {

core/src/consensus.rs

@@ -32,7 +32,6 @@ use std::{
         Deref,
     },
     path::{Path, PathBuf},
-    sync::Arc,
 };
 use thiserror::Error;
 
@@ -1190,7 +1189,7 @@ impl Tower {
         path.with_extension("bin.new")
     }
 
-    pub fn save(&self, node_keypair: &Arc<Keypair>) -> Result<()> {
+    pub fn save(&self, node_keypair: &Keypair) -> Result<()> {
         let mut measure = Measure::start("tower_save-ms");
 
         if self.node_pubkey != node_keypair.pubkey() {
@@ -1293,7 +1292,7 @@ pub struct SavedTower {
 }
 
 impl SavedTower {
-    pub fn new<T: Signer>(tower: &Tower, keypair: &Arc<T>) -> Result<Self> {
+    pub fn new<T: Signer>(tower: &Tower, keypair: &T) -> Result<Self> {
         let data = bincode::serialize(tower)?;
         let signature = keypair.sign_message(&data);
         Ok(Self { signature, data })
@@ -1391,7 +1390,7 @@ pub mod test {
         collections::HashMap,
         fs::{remove_file, OpenOptions},
         io::{Read, Seek, SeekFrom, Write},
-        sync::RwLock,
+        sync::{Arc, RwLock},
     };
     use tempfile::TempDir;
     use trees::{tr, Tree, TreeWalk};

core/src/replay_stage.rs

@@ -339,7 +339,8 @@ impl ReplayStage {
             .spawn(move || {
                 let verify_recyclers = VerifyRecyclers::default();
                 let _exit = Finalizer::new(exit.clone());
-                let my_pubkey = cluster_info.id();
+                let mut identity_keypair = cluster_info.keypair().clone();
+                let mut my_pubkey = identity_keypair.pubkey();
                 let (
                     mut progress,
                     mut heaviest_subtree_fork_choice,
@@ -515,6 +516,7 @@ impl ReplayStage {
                                                     heaviest_bank_on_same_voted_fork,
                                                     &poh_recorder, my_latest_landed_vote,
                                                     &vote_account,
+                                                    &identity_keypair,
                                                     &authorized_voter_keypairs.read().unwrap(),
                                                     &mut voted_signatures,
                                                     has_new_vote_been_rooted, &mut
@@ -582,6 +584,7 @@ impl ReplayStage {
                             &mut tower,
                             &mut progress,
                             &vote_account,
+                            &identity_keypair,
                             &authorized_voter_keypairs.read().unwrap(),
                             &cluster_info,
                             &blockstore,
@@ -627,6 +630,14 @@ impl ReplayStage {
                                     i64
                                 ),
                             );
+
+                            if my_pubkey != cluster_info.id() {
+                                identity_keypair = cluster_info.keypair().clone();
+                                let my_old_pubkey = my_pubkey;
+                                my_pubkey = identity_keypair.pubkey();
+                                warn!("Identity changed from {} to {}", my_old_pubkey, my_pubkey);
+                            }
+
                             Self::reset_poh_recorder(
                                 &my_pubkey,
                                 &blockstore,
@@ -1290,6 +1301,7 @@ impl ReplayStage {
         tower: &mut Tower,
         progress: &mut ProgressMap,
         vote_account_pubkey: &Pubkey,
+        identity_keypair: &Keypair,
         authorized_voter_keypairs: &[Arc<Keypair>],
         cluster_info: &Arc<ClusterInfo>,
         blockstore: &Arc<Blockstore>,
@@ -1314,7 +1326,7 @@ impl ReplayStage {
         trace!("handle votable bank {}", bank.slot());
         let new_root = tower.record_bank_vote(bank, vote_account_pubkey);
 
-        if let Err(err) = tower.save(&cluster_info.keypair) {
+        if let Err(err) = tower.save(identity_keypair) {
             error!("Unable to save tower: {:?}", err);
             std::process::exit(1);
         }
@@ -1386,6 +1398,7 @@ impl ReplayStage {
             bank,
             poh_recorder,
             vote_account_pubkey,
+            identity_keypair,
             authorized_voter_keypairs,
             tower,
             switch_fork_decision,
@@ -1396,7 +1409,7 @@ impl ReplayStage {
     }
 
     fn generate_vote_tx(
-        node_keypair: &Arc<Keypair>,
+        node_keypair: &Keypair,
         bank: &Bank,
         vote_account_pubkey: &Pubkey,
         authorized_voter_keypairs: &[Arc<Keypair>],
@@ -1465,7 +1478,7 @@ impl ReplayStage {
         let mut vote_tx = Transaction::new_with_payer(&[vote_ix], Some(&node_keypair.pubkey()));
 
         let blockhash = bank.last_blockhash();
-        vote_tx.partial_sign(&[node_keypair.as_ref()], blockhash);
+        vote_tx.partial_sign(&[node_keypair], blockhash);
         vote_tx.partial_sign(&[authorized_voter_keypair.as_ref()], blockhash);
 
         if !has_new_vote_been_rooted {
@@ -1488,6 +1501,7 @@ impl ReplayStage {
         poh_recorder: &Mutex<PohRecorder>,
         my_latest_landed_vote: Slot,
         vote_account_pubkey: &Pubkey,
+        identity_keypair: &Keypair,
         authorized_voter_keypairs: &[Arc<Keypair>],
         vote_signatures: &mut Vec<Signature>,
         has_new_vote_been_rooted: bool,
@@ -1526,7 +1540,7 @@ impl ReplayStage {
         // TODO: check the timestamp in this vote is correct, i.e. it shouldn't
         // have changed from the original timestamp of the vote.
         let vote_tx = Self::generate_vote_tx(
-            &cluster_info.keypair,
+            identity_keypair,
             heaviest_bank_on_same_fork,
             vote_account_pubkey,
             authorized_voter_keypairs,
@@ -1563,6 +1577,7 @@ impl ReplayStage {
         bank: &Bank,
         poh_recorder: &Mutex<PohRecorder>,
         vote_account_pubkey: &Pubkey,
+        identity_keypair: &Keypair,
         authorized_voter_keypairs: &[Arc<Keypair>],
         tower: &mut Tower,
         switch_fork_decision: &SwitchForkDecision,
@@ -1572,7 +1587,7 @@ impl ReplayStage {
     ) {
         let mut generate_time = Measure::start("generate_vote");
         let vote_tx = Self::generate_vote_tx(
-            &cluster_info.keypair,
+            identity_keypair,
             bank,
             vote_account_pubkey,
             authorized_voter_keypairs,
@@ -4668,6 +4683,7 @@ mod tests {
         let has_new_vote_been_rooted = false;
         let mut voted_signatures = vec![];
 
+        let identity_keypair = cluster_info.keypair().clone();
         let my_vote_keypair = vec![Arc::new(
             validator_keypairs.remove(&my_pubkey).unwrap().vote_keypair,
         )];
@@ -4693,6 +4709,7 @@ mod tests {
             &bank0,
             &poh_recorder,
             &my_vote_pubkey,
+            &identity_keypair,
             &my_vote_keypair,
             &mut tower,
             &SwitchForkDecision::SameFork,
@@ -4723,6 +4740,7 @@ mod tests {
                 &poh_recorder,
                 Tower::last_voted_slot_in_bank(refresh_bank, &my_vote_pubkey).unwrap(),
                 &my_vote_pubkey,
+                &identity_keypair,
                 &my_vote_keypair,
                 &mut voted_signatures,
                 has_new_vote_been_rooted,
@@ -4745,6 +4763,7 @@ mod tests {
             &bank1,
             &poh_recorder,
             &my_vote_pubkey,
+            &identity_keypair,
             &my_vote_keypair,
             &mut tower,
             &SwitchForkDecision::SameFork,
@@ -4768,6 +4787,7 @@ mod tests {
             &poh_recorder,
             Tower::last_voted_slot_in_bank(&bank2, &my_vote_pubkey).unwrap(),
             &my_vote_pubkey,
+            &identity_keypair,
             &my_vote_keypair,
             &mut voted_signatures,
             has_new_vote_been_rooted,
@@ -4804,6 +4824,7 @@ mod tests {
             &poh_recorder,
             Tower::last_voted_slot_in_bank(&expired_bank, &my_vote_pubkey).unwrap(),
             &my_vote_pubkey,
+            &identity_keypair,
             &my_vote_keypair,
             &mut voted_signatures,
             has_new_vote_been_rooted,
@@ -4860,6 +4881,7 @@ mod tests {
             &poh_recorder,
             Tower::last_voted_slot_in_bank(&expired_bank_sibling, &my_vote_pubkey).unwrap(),
             &my_vote_pubkey,
+            &identity_keypair,
             &my_vote_keypair,
             &mut voted_signatures,
             has_new_vote_been_rooted,

core/src/test_validator.rs

@@ -1,7 +1,11 @@
 use {
     crate::validator::{Validator, ValidatorConfig, ValidatorStartProgress},
     solana_client::rpc_client::RpcClient,
-    solana_gossip::{cluster_info::Node, gossip_service::discover_cluster, socketaddr},
+    solana_gossip::{
+        cluster_info::{ClusterInfo, Node},
+        gossip_service::discover_cluster,
+        socketaddr,
+    },
     solana_ledger::{blockstore::create_new_ledger, create_new_tmp_ledger},
     solana_net_utils::PortRange,
     solana_rpc::rpc::JsonRpcConfig,
@@ -502,7 +506,7 @@ impl TestValidator {
 
         let validator = Some(Validator::new(
             node,
-            &Arc::new(validator_identity),
+            Arc::new(validator_identity),
             &ledger_path,
             &vote_account_address,
             config.authorized_voter_keypairs.clone(),
@@ -599,6 +603,10 @@ impl TestValidator {
             validator.join();
         }
     }
+
+    pub fn cluster_info(&self) -> Arc<ClusterInfo> {
+        self.validator.as_ref().unwrap().cluster_info.clone()
+    }
 }
 
 impl Drop for TestValidator {

core/src/validator.rs

@@ -260,6 +260,7 @@ pub struct Validator {
     tpu: Tpu,
     tvu: Tvu,
     ip_echo_server: Option<solana_net_utils::IpEchoServer>,
+    pub cluster_info: Arc<ClusterInfo>,
 }
 
 // in the distant future, get rid of ::new()/exit() and use Result properly...
@@ -279,7 +280,7 @@ pub(crate) fn abort() -> ! {
 impl Validator {
     pub fn new(
         mut node: Node,
-        identity_keypair: &Arc<Keypair>,
+        identity_keypair: Arc<Keypair>,
         ledger_path: &Path,
         vote_account: &Pubkey,
         authorized_voter_keypairs: Arc<RwLock<Vec<Arc<Keypair>>>>,
@@ -437,7 +438,7 @@ impl Validator {
             }
         }
 
-        let mut cluster_info = ClusterInfo::new(node.info.clone(), identity_keypair.clone());
+        let mut cluster_info = ClusterInfo::new(node.info.clone(), identity_keypair);
         cluster_info.set_contact_debug_interval(config.contact_debug_interval);
         cluster_info.set_entrypoints(cluster_entrypoints);
         cluster_info.restore_contact_info(ledger_path, config.contact_save_interval);
@@ -790,6 +791,7 @@ impl Validator {
             poh_recorder,
             ip_echo_server,
             validator_exit: config.validator_exit.clone(),
+            cluster_info,
         }
     }
 
@@ -829,6 +831,8 @@ impl Validator {
     }
 
     pub fn join(self) {
+        drop(self.cluster_info);
+
         self.poh_service.join().expect("poh_service");
         drop(self.poh_recorder);
 
@@ -1628,7 +1632,7 @@ mod tests {
         let start_progress = Arc::new(RwLock::new(ValidatorStartProgress::default()));
         let validator = Validator::new(
             validator_node,
-            &Arc::new(validator_keypair),
+            Arc::new(validator_keypair),
             &validator_ledger_path,
             &voting_keypair.pubkey(),
             Arc::new(RwLock::new(vec![voting_keypair.clone()])),
@@ -1706,7 +1710,7 @@ mod tests {
                 };
                 Validator::new(
                     validator_node,
-                    &Arc::new(validator_keypair),
+                    Arc::new(validator_keypair),
                     &validator_ledger_path,
                     &vote_account_keypair.pubkey(),
                     Arc::new(RwLock::new(vec![Arc::new(vote_account_keypair)])),