Merge pull request #835 from openmina/fix/transition_frontier/use_random_peers_for_fetch

Transition Frontier Sync: use random peer to fetch staged ledger parts or best tip

Author: binier

Cargo.lock

@@ -5,6 +5,7 @@ edition = "2021"
 license = "Apache-2.0"
 
 [dependencies]
+rand = "0.8.0"
 serde = "1.0.147"
 serde_json = { version = "1.0.82", features = ["unbounded_depth", "arbitrary_precision"] }
 serde_with = { version = "3.6.1", features = ["time_0_3"] }

node/Cargo.toml

@@ -1,4 +1,5 @@
 use openmina_core::log::system_time;
+use rand::prelude::*;
 
 use crate::block_producer::{block_producer_effects, BlockProducerAction};
 use crate::event_source::event_source_effects;
@@ -120,10 +121,12 @@ use mina_p2p_messages::v2::StateHash;
 
 fn request_best_tip<S: Service>(store: &mut Store<S>, _consensus_best_tip_hash: Option<StateHash>) {
     let p2p = p2p_ready!(store.state().p2p, "request_best_tip", system_time());
-    if let Some((peer_id, id)) = p2p.ready_rpc_peers_iter().last() {
+
+    let peers = p2p.ready_rpc_peers_iter().collect::<Vec<_>>();
+    if let Some((peer_id, id)) = peers.choose(&mut store.state().pseudo_rng()) {
         store.dispatch(P2pChannelsRpcAction::RequestSend {
-            peer_id,
-            id,
+            peer_id: *peer_id,
+            id: *id,
             request: Box::new(P2pRpcRequest::BestTipWithProof),
             on_init: None,
         });

node/src/effects.rs

@@ -1,6 +1,7 @@
 use std::sync::Arc;
 
 use mina_p2p_messages::v2::{MinaBaseUserCommandStableV2, MinaBlockBlockStableV2};
+use rand::prelude::*;
 
 use openmina_core::block::BlockWithHash;
 use openmina_core::requests::RpcId;
@@ -252,6 +253,10 @@ impl State {
         self.last_action.time()
     }
 
+    pub fn pseudo_rng(&self) -> StdRng {
+        StdRng::seed_from_u64(self.time().into())
+    }
+
     /// Must be called in the global reducer as the last thing only once
     /// and only there!
     pub fn action_applied(&mut self, action: &ActionWithMeta) {

node/src/state.rs

@@ -6,6 +6,7 @@ use p2p::{
     },
     PeerId,
 };
+use rand::prelude::*;
 
 use crate::ledger::write::{LedgerWriteAction, LedgerWriteRequest};
 
@@ -44,9 +45,10 @@ impl TransitionFrontierSyncLedgerStagedState {
                 };
                 let block_hash = staged_ledger.target().staged.block_hash.clone();
 
-                let ready_peers = staged_ledger
+                let mut ready_peers = staged_ledger
                     .filter_available_peers(p2p.ready_rpc_peers_iter())
                     .collect::<Vec<_>>();
+                ready_peers.shuffle(&mut global_state.pseudo_rng());
 
                 for (peer_id, rpc_id) in ready_peers {
                     let enqueued = if p2p.is_libp2p_peer(&peer_id) {

node/src/transition_frontier/sync/ledger/staged/transition_frontier_sync_ledger_staged_reducer.rs

@@ -52,7 +52,7 @@ impl crate::P2pState {
         );
 
         /// random shuffle available peers
-        use rand::{seq::SliceRandom, SeedableRng};
+        use rand::prelude::*;
         let mut rng = rand::rngs::StdRng::seed_from_u64(time.into());
         let mut available_peers_ordered = available_peers.iter().copied().collect::<Vec<_>>();
         available_peers_ordered.shuffle(&mut rng);