wen_restart: correctly handle HeaviestFork received from the coordina…

…tor. (#2923)

* wen_restart: Change HeaviestFork stage to a leader based approach.

* Fix logic to calculate slots to repair, add more logs.

* Let the leader generate snapshot and print error log before continuing to send out heaviest fork.

* Filter ancestors older than root.

* Reduce lock scope.

* Rename variables and functions.

* Make leader aggregate the heaviest fork of everyone.

* Move heaviest fork aggregation to DONE stage.

* No warning when receiving HeaviestFork from non-leader.

* Add test for receive_restart_heaviest_fork.

* Add test for repair_heaviest_fork.

* Add test for verify_leader_heaviest_fork.

* Rename wen_restart_leader to wen_restart_coordinator.

* Fix a bad merge

* Non-coordinator should use coodinator's (slot, hash) after verification.

* Remove trailing whitespace.

* Fix a bad merge

* Fix a bad merge.

* Remove unnecessary changes.

* Make coordinator select different slot in test, wait before exiting with error.

* Add send_and_receive_heaviest_fork and test.

* Make the coordinator print stats every 10 seconds.

* Rename variables and add comments.

* Rename final_restart_slot/hash with my_heaviest_fork_slot/hash

* flush_push_queue before waiting to speed things up.

core/src/validator.rs

@@ -1433,7 +1433,7 @@ impl Validator {
         .map_err(ValidatorError::Other)?;
 
         if in_wen_restart {
-            info!("Waiting for wen_restart phase one to finish");
+            info!("Waiting for wen_restart to finish");
             wait_for_wen_restart(WenRestartConfig {
                 wen_restart_path: config.wen_restart_proto_path.clone().unwrap(),
                 wen_restart_coordinator: config.wen_restart_coordinator.unwrap(),

wen-restart/src/heaviest_fork_aggregate.rs

@@ -161,8 +161,10 @@ impl HeaviestForkAggregate {
         })
     }
 
-    pub(crate) fn block_stake_map(self) -> HashMap<(Slot, Hash), u64> {
-        self.block_stake_map
+    pub(crate) fn print_block_stake_map(&self) {
+        for ((slot, hash), stake) in self.block_stake_map.iter() {
+            info!("Slot: {}, Hash: {}, Stake: {}", slot, hash, stake,);
+        }
     }
 }