Merge pull request #978 from openmina/feat/better_seed_node

Better seed node

Author: binier

Cargo.lock

@@ -75,6 +75,9 @@ pub struct Node {
     #[arg(long, env)]
     pub peer_list_url: Option<Url>,
 
+    #[arg(long, default_value = "100")]
+    pub max_peers: usize,
+
     /// Run the node in seed mode. No default peers will be added.
     #[arg(long, env)]
     pub seed: bool,
@@ -206,6 +209,7 @@ impl Node {
                 .filter_map(|s| s.parse().ok()),
         );
 
+        node_builder.p2p_max_peers(self.max_peers);
         self.seed.then(|| node_builder.p2p_seed_node());
         self.no_peers_discovery
             .then(|| node_builder.p2p_no_discovery());

cli/src/commands/node/mod.rs

@@ -14,6 +14,7 @@ tracing = { version = "0.1", features = ["std"] }
 sha2 = "0.10.6"
 binprot = { git = "https://github.com/openmina/binprot-rs", rev = "2b5a909" }
 binprot_derive = { git = "https://github.com/openmina/binprot-rs", rev = "2b5a909" }
+rand = "0.8.0"
 redux = { workspace = true }
 tokio = { version = "1.26", features = ["sync"] }
 time = { version = "0.3", features = ["formatting", "macros", "parsing"] }

core/Cargo.toml

@@ -52,6 +52,12 @@ mod work_dir {
 
 pub use work_dir::{get_debug_dir, get_work_dir, set_work_dir};
 
+use rand::prelude::*;
+#[inline(always)]
+pub fn pseudo_rng(time: redux::Timestamp) -> StdRng {
+    StdRng::seed_from_u64(time.into())
+}
+
 pub fn preshared_key(chain_id: &ChainId) -> [u8; 32] {
     use multihash::Hasher;
     let mut hasher = Blake2b256::default();

core/src/lib.rs

@@ -4,6 +4,7 @@ use std::{
     net::IpAddr,
     path::Path,
     sync::Arc,
+    time::Duration,
 };
 
 use anyhow::Context;
@@ -34,13 +35,10 @@ pub struct NodeBuilder {
     rng_seed: [u8; 32],
     custom_initial_time: Option<redux::Timestamp>,
     genesis_config: Arc<GenesisConfig>,
+    p2p: P2pConfig,
     p2p_sec_key: Option<P2pSecretKey>,
-    p2p_libp2p_port: Option<u16>,
     p2p_is_seed: bool,
-    p2p_no_discovery: bool,
     p2p_is_started: bool,
-    initial_peers: Vec<P2pConnectionOutgoingInitOpts>,
-    external_addrs: Vec<IpAddr>,
     block_producer: Option<BlockProducerConfig>,
     snarker: Option<SnarkerConfig>,
     service: NodeServiceBuilder,
@@ -68,13 +66,25 @@ impl NodeBuilder {
             rng_seed,
             custom_initial_time: None,
             genesis_config,
+            p2p: P2pConfig {
+                libp2p_port: None,
+                listen_port: None,
+                // Must be replaced with builder api.
+                identity_pub_key: P2pSecretKey::deterministic(0).public_key(),
+                initial_peers: Vec::new(),
+                external_addrs: Vec::new(),
+                enabled_channels: ChannelId::iter_all().collect(),
+                peer_discovery: true,
+                meshsub: P2pMeshsubConfig {
+                    initial_time: Duration::ZERO,
+                    ..Default::default()
+                },
+                timeouts: P2pTimeouts::default(),
+                limits: P2pLimits::default().with_max_peers(Some(100)),
+            },
             p2p_sec_key: None,
-            p2p_libp2p_port: None,
             p2p_is_seed: false,
-            p2p_no_discovery: false,
             p2p_is_started: false,
-            initial_peers: Vec::new(),
-            external_addrs: Vec::new(),
             block_producer: None,
             snarker: None,
             service: NodeServiceBuilder::new(rng_seed),
@@ -94,12 +104,13 @@ impl NodeBuilder {
 
     /// If not called, random one will be generated and used instead.
     pub fn p2p_sec_key(&mut self, key: P2pSecretKey) -> &mut Self {
+        self.p2p.identity_pub_key = key.public_key();
         self.p2p_sec_key = Some(key);
         self
     }
 
     pub fn p2p_libp2p_port(&mut self, port: u16) -> &mut Self {
-        self.p2p_libp2p_port = Some(port);
+        self.p2p.libp2p_port = Some(port);
         self
     }
 
@@ -110,7 +121,7 @@ impl NodeBuilder {
     }
 
     pub fn p2p_no_discovery(&mut self) -> &mut Self {
-        self.p2p_no_discovery = true;
+        self.p2p.peer_discovery = false;
         self
     }
 
@@ -119,19 +130,19 @@ impl NodeBuilder {
         &mut self,
         peers: impl IntoIterator<Item = P2pConnectionOutgoingInitOpts>,
     ) -> &mut Self {
-        self.initial_peers.extend(peers);
+        self.p2p.initial_peers.extend(peers);
         self
     }
 
     pub fn external_addrs(&mut self, v: impl Iterator<Item = IpAddr>) -> &mut Self {
-        self.external_addrs.extend(v);
+        self.p2p.external_addrs.extend(v);
         self
     }
 
     /// Extend p2p initial peers from file.
     pub fn initial_peers_from_file(&mut self, path: impl AsRef<Path>) -> anyhow::Result<&mut Self> {
         peers_from_reader(
-            &mut self.initial_peers,
+            &mut self.p2p.initial_peers,
             File::open(&path).context(anyhow::anyhow!(
                 "opening peer list file {:?}",
                 path.as_ref()
@@ -152,14 +163,19 @@ impl NodeBuilder {
     ) -> anyhow::Result<&mut Self> {
         let url = url.into_url().context("failed to parse peers url")?;
         peers_from_reader(
-            &mut self.initial_peers,
+            &mut self.p2p.initial_peers,
             reqwest::blocking::get(url.clone())
                 .context(anyhow::anyhow!("reading peer list url {url}"))?,
         )
         .context(anyhow::anyhow!("reading peer list url {url}"))?;
         Ok(self)
     }
 
+    pub fn p2p_max_peers(&mut self, limit: usize) -> &mut Self {
+        self.p2p.limits = self.p2p.limits.with_max_peers(Some(limit));
+        self
+    }
+
     /// Override default p2p task spawner.
     pub fn p2p_custom_task_spawner(
         &mut self,
@@ -274,15 +290,15 @@ impl NodeBuilder {
         self
     }
 
-    pub fn build(self) -> anyhow::Result<Node> {
+    pub fn build(mut self) -> anyhow::Result<Node> {
         let p2p_sec_key = self.p2p_sec_key.unwrap_or_else(P2pSecretKey::rand);
-        let initial_peers = if self.initial_peers.is_empty() && !self.p2p_is_seed {
-            default_peers()
-        } else {
-            self.initial_peers
-        };
+        if self.p2p.initial_peers.is_empty() && !self.p2p_is_seed {
+            self.p2p.initial_peers = default_peers();
+        }
 
-        let initial_peers = initial_peers
+        self.p2p.initial_peers = self
+            .p2p
+            .initial_peers
             .into_iter()
             .filter_map(|opts| match opts {
                 P2pConnectionOutgoingInitOpts::LibP2P(mut opts) => {
@@ -293,8 +309,6 @@ impl NodeBuilder {
             })
             .collect();
 
-        let external_addrs = self.external_addrs;
-
         let srs = self.verifier_srs.unwrap_or_else(get_srs);
         let block_verifier_index = self
             .block_verifier_index
@@ -306,6 +320,9 @@ impl NodeBuilder {
         let initial_time = self
             .custom_initial_time
             .unwrap_or_else(redux::Timestamp::global_now);
+        self.p2p.meshsub.initial_time = initial_time
+            .checked_sub(redux::Timestamp::ZERO)
+            .unwrap_or_default();
 
         let protocol_constants = self.genesis_config.protocol_constants()?;
         let consensus_consts =
@@ -319,23 +336,7 @@ impl NodeBuilder {
                 consensus_constants: consensus_consts.clone(),
                 testing_run: false,
             },
-            p2p: P2pConfig {
-                libp2p_port: self.p2p_libp2p_port,
-                listen_port: self.http_port,
-                identity_pub_key: p2p_sec_key.public_key(),
-                initial_peers,
-                external_addrs,
-                enabled_channels: ChannelId::iter_all().collect(),
-                peer_discovery: !self.p2p_no_discovery,
-                meshsub: P2pMeshsubConfig {
-                    initial_time: initial_time
-                        .checked_sub(redux::Timestamp::ZERO)
-                        .unwrap_or_default(),
-                    ..Default::default()
-                },
-                timeouts: P2pTimeouts::default(),
-                limits: P2pLimits::default().with_max_peers(Some(100)),
-            },
+            p2p: self.p2p,
             ledger: LedgerConfig {},
             snark: SnarkConfig {
                 block_verifier_index,

node/native/src/node/builder.rs

@@ -343,6 +343,7 @@ pub enum ActionKind {
     P2pDisconnectionFinish,
     P2pDisconnectionInit,
     P2pDisconnectionPeerClosed,
+    P2pDisconnectionRandomTry,
     P2pDisconnectionEffectfulInit,
     P2pEffectfulInitialize,
     P2pIdentifyNewRequest,
@@ -705,7 +706,7 @@ pub enum ActionKind {
 }
 
 impl ActionKind {
-    pub const COUNT: u16 = 595;
+    pub const COUNT: u16 = 596;
 }
 
 impl std::fmt::Display for ActionKind {
@@ -1212,6 +1213,7 @@ impl ActionKindGet for P2pConnectionAction {
 impl ActionKindGet for P2pDisconnectionAction {
     fn kind(&self) -> ActionKind {
         match self {
+            Self::RandomTry => ActionKind::P2pDisconnectionRandomTry,
             Self::Init { .. } => ActionKind::P2pDisconnectionInit,
             Self::PeerClosed { .. } => ActionKind::P2pDisconnectionPeerClosed,
             Self::FailedCleanup { .. } => ActionKind::P2pDisconnectionFailedCleanup,

node/src/action_kind.rs

@@ -309,7 +309,7 @@ impl State {
     }
 
     pub fn pseudo_rng(&self) -> StdRng {
-        StdRng::seed_from_u64(self.time().into())
+        crate::core::pseudo_rng(self.time())
     }
 
     /// Must be called in the global reducer as the last thing only once

node/src/state.rs

@@ -15,6 +15,8 @@ use crate::{
 
 #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, thiserror::Error)]
 pub enum P2pDisconnectionReason {
+    #[error("random disconnection in order to free up space so that more peers can connect")]
+    FreeUpSpace,
     #[error("message is unexpected for channel {0}")]
     P2pChannelMsgUnexpected(ChannelId),
     #[error("failed to send message to channel: {0}")]

p2p/src/disconnection/mod.rs

@@ -1,3 +1,5 @@
+use std::time::Duration;
+
 use openmina_core::ActionEvent;
 use serde::{Deserialize, Serialize};
 
@@ -6,9 +8,12 @@ use crate::{P2pPeerStatus, P2pState, PeerId};
 
 pub type P2pDisconnectionActionWithMetaRef<'a> = redux::ActionWithMeta<&'a P2pDisconnectionAction>;
 
+const RANDOM_DISCONNECTION_TRY_FREQUENCY: Duration = Duration::from_secs(10);
+
 #[derive(Serialize, Deserialize, Debug, Clone, ActionEvent)]
 #[action_event(level = debug)]
 pub enum P2pDisconnectionAction {
+    RandomTry,
     /// Initialize disconnection.
     #[action_event(fields(display(peer_id), display(reason)), level = info)]
     Init {
@@ -17,17 +22,26 @@ pub enum P2pDisconnectionAction {
     },
     /// Peer disconnection.
     #[action_event(fields(display(peer_id)), level = info)]
-    PeerClosed { peer_id: PeerId },
+    PeerClosed {
+        peer_id: PeerId,
+    },
     #[action_event(fields(display(peer_id)), level = info)]
-    FailedCleanup { peer_id: PeerId },
+    FailedCleanup {
+        peer_id: PeerId,
+    },
     /// Finish disconnecting from a peer.
     #[action_event(fields(display(peer_id)), level = debug)]
-    Finish { peer_id: PeerId },
+    Finish {
+        peer_id: PeerId,
+    },
 }
 
 impl redux::EnablingCondition<P2pState> for P2pDisconnectionAction {
-    fn is_enabled(&self, state: &P2pState, _time: redux::Timestamp) -> bool {
+    fn is_enabled(&self, state: &P2pState, time: redux::Timestamp) -> bool {
         match self {
+            P2pDisconnectionAction::RandomTry => time
+                .checked_sub(state.last_random_disconnection_try)
+                .map_or(false, |dur| dur >= RANDOM_DISCONNECTION_TRY_FREQUENCY),
             P2pDisconnectionAction::Init { peer_id, .. }
             | P2pDisconnectionAction::PeerClosed { peer_id, .. }
             | P2pDisconnectionAction::Finish { peer_id } => {

p2p/src/disconnection/p2p_disconnection_actions.rs

@@ -1,12 +1,18 @@
-use openmina_core::{bug_condition, Substate};
+use std::time::Duration;
+
+use openmina_core::{bug_condition, pseudo_rng, Substate};
+use rand::prelude::*;
 use redux::ActionWithMeta;
 
 use crate::{
     disconnection_effectful::P2pDisconnectionEffectfulAction, P2pNetworkSchedulerAction,
     P2pPeerAction, P2pPeerStatus, P2pState,
 };
 
-use super::{P2pDisconnectedState, P2pDisconnectionAction};
+use super::{P2pDisconnectedState, P2pDisconnectionAction, P2pDisconnectionReason};
+
+/// Do not disconnect peer for this duration just for freeing up peer space.
+const FORCE_PEER_STABLE_FOR: Duration = Duration::from_secs(90);
 
 impl P2pDisconnectedState {
     pub fn reducer<Action, State>(
@@ -21,6 +27,30 @@ impl P2pDisconnectedState {
         let p2p_state = state_context.get_substate_mut()?;
 
         match action {
+            P2pDisconnectionAction::RandomTry => {
+                p2p_state.last_random_disconnection_try = meta.time();
+                if p2p_state.config.limits.max_stable_peers()
+                    >= p2p_state.ready_peers_iter().count()
+                {
+                    return Ok(());
+                }
+                let mut rng = pseudo_rng(meta.time());
+
+                let peer_id = p2p_state
+                    .ready_peers_iter()
+                    .filter(|(_, s)| s.connected_for(meta.time()) > FORCE_PEER_STABLE_FOR)
+                    .map(|(id, _)| *id)
+                    .choose(&mut rng);
+
+                if let Some(peer_id) = peer_id {
+                    let dispatcher = state_context.into_dispatcher();
+                    dispatcher.push(P2pDisconnectionAction::Init {
+                        peer_id,
+                        reason: P2pDisconnectionReason::FreeUpSpace,
+                    });
+                }
+                Ok(())
+            }
             P2pDisconnectionAction::Init { peer_id, reason } => {
                 #[cfg(feature = "p2p-libp2p")]
                 if p2p_state.is_libp2p_peer(&peer_id) {