Merge pull request #1009 from openmina/fix/p2p/webrtc/seed_peers_gone

P2P WebRTC Fixes

Author: binier

core/src/log.rs

@@ -164,7 +164,7 @@ pub use crate::{debug, error, info, trace, warn};
 macro_rules! bug_condition {
     ($($arg:tt)*) => {{
         if std::env::var("OPENMINA_PANIC_ON_BUG")
-        .map(|v| v.to_lowercase() == "true")
+        .map(|v| ["true", "1"].contains(&v.to_lowercase().as_str()))
         .unwrap_or(false) {
             panic!($($arg)*)
         } else {

node/native/src/node/builder.rs

@@ -301,6 +301,7 @@ impl NodeBuilder {
             .p2p
             .initial_peers
             .into_iter()
+            .filter(|opts| *opts.peer_id() != p2p_sec_key.public_key().peer_id())
             .filter_map(|opts| match opts {
                 P2pConnectionOutgoingInitOpts::LibP2P(mut opts) => {
                     opts.host = opts.host.resolve()?;

node/testing/src/node/rust/mod.rs

@@ -187,7 +187,7 @@ impl Node {
         false
     }
 
-    pub fn p2p_disconnect(&mut self, peer_id: PeerId) {
+    pub fn p2p_disconnect(&mut self, peer_id: PeerId) -> bool {
         self.service_mut().disconnect(peer_id)
     }
 }

node/testing/src/scenarios/record_replay/block_production.rs

@@ -35,7 +35,7 @@ impl RecordReplayBlockProduction {
             block_producers: 3,
             advance_time: RunCfgAdvanceTime::Rand(1..=200),
             run_until: SimulatorRunUntil::BlockchainLength(10),
-            run_until_timeout: Duration::from_secs(10 * 60),
+            run_until_timeout: Duration::from_secs(15 * 60),
             recorder: Recorder::StateWithInputActions,
         };
         let mut simulator = Simulator::new(initial_time, cfg);

node/web/src/node/builder.rs

@@ -202,6 +202,10 @@ impl NodeBuilder {
         } else {
             self.initial_peers
         };
+        let initial_peers = initial_peers
+            .into_iter()
+            .filter(|opts| *opts.peer_id() != p2p_sec_key.public_key().peer_id())
+            .collect();
 
         let srs = self.verifier_srs.unwrap_or_else(get_srs);
         let block_verifier_index = self

p2p/src/disconnection/p2p_disconnection_reducer.rs

@@ -123,7 +123,18 @@ impl P2pDisconnectedState {
 
                 let (dispatcher, state) = state_context.into_dispatcher_and_state();
                 let p2p_state: &P2pState = state.substate()?;
-                dispatcher.push(P2pPeerAction::Remove { peer_id });
+
+                // remove oldest disconnected peer
+                if let Some((peer_id, _)) = p2p_state
+                    .peers
+                    .iter()
+                    .filter_map(|(id, p)| {
+                        Some((*id, p.status.disconnected_or_disconnecting_time()?))
+                    })
+                    .min_by_key(|(_, t)| *t)
+                {
+                    dispatcher.push(P2pPeerAction::Remove { peer_id });
+                }
 
                 if let Some(callback) = &p2p_state.callbacks.on_p2p_disconnection_finish {
                     dispatcher.push_callback(callback.clone(), peer_id);

p2p/src/disconnection_effectful/p2p_disconnection_effectful_effects.rs

@@ -1,5 +1,7 @@
 use redux::ActionMeta;
 
+use crate::disconnection::P2pDisconnectionAction;
+
 use super::{P2pDisconnectionEffectfulAction, P2pDisconnectionService};
 
 impl P2pDisconnectionEffectfulAction {
@@ -10,7 +12,10 @@ impl P2pDisconnectionEffectfulAction {
     {
         match self {
             P2pDisconnectionEffectfulAction::Init { peer_id } => {
-                store.service().disconnect(peer_id);
+                if store.service().disconnect(peer_id) {
+                    // peer was already disconnected, so dispatch finish.
+                    store.dispatch(P2pDisconnectionAction::Finish { peer_id });
+                }
             }
         }
     }

p2p/src/disconnection_effectful/p2p_disconnection_effectful_service.rs

@@ -1,5 +1,5 @@
 use crate::PeerId;
 
 pub trait P2pDisconnectionService: redux::Service {
-    fn disconnect(&mut self, peer_id: PeerId);
+    fn disconnect(&mut self, peer_id: PeerId) -> bool;
 }

p2p/src/p2p_config.rs

@@ -102,7 +102,7 @@ impl Default for P2pTimeouts {
                 "OUTGOING_CONNECTION_TIMEOUT",
                 Some(Duration::from_secs(15)),
             ),
-            reconnect_timeout: from_env_or("RECONNECT_TIMEOUT", Some(Duration::from_secs(1))),
+            reconnect_timeout: from_env_or("RECONNECT_TIMEOUT", Some(Duration::from_secs(30))),
             incoming_error_reconnect_timeout: from_env_or(
                 "INCOMING_ERROR_RECONNECT_TIMEOUT",
                 Some(Duration::from_secs(30)),

p2p/src/p2p_reducer.rs

@@ -83,6 +83,7 @@ impl P2pState {
         dispatcher.push(P2pConnectionOutgoingAction::RandomInit);
         dispatcher.push(P2pDisconnectionAction::RandomTry);
 
+        state.p2p_connect_initial_peers(dispatcher);
         state.p2p_try_reconnect_disconnected_peers(dispatcher, time)?;
         state.p2p_discovery(dispatcher, time)?;
 
@@ -153,6 +154,29 @@ impl P2pState {
         Ok(())
     }
 
+    fn p2p_connect_initial_peers<State, Action>(&self, dispatcher: &mut Dispatcher<Action, State>)
+    where
+        State: crate::P2pStateTrait,
+        Action: crate::P2pActionTrait<State>,
+    {
+        if self.ready_peers_iter().count() >= self.config.initial_peers.len() {
+            return;
+        }
+
+        self.config
+            .initial_peers
+            .iter()
+            .filter(|opts| !self.peers.contains_key(opts.peer_id()))
+            .cloned()
+            .for_each(|opts| {
+                dispatcher.push(P2pConnectionOutgoingAction::Init {
+                    opts,
+                    rpc_id: None,
+                    on_success: None,
+                });
+            });
+    }
+
     fn rpc_timeouts<State, Action>(
         &self,
         dispatcher: &mut Dispatcher<Action, State>,