refactor: Improve rendezvous protocol usage

homestar-runtime/fixtures/__testkey_ed25519_2.pem

@@ -0,0 +1,3 @@
+-----BEGIN PRIVATE KEY-----
+AQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQE=
+-----END PRIVATE KEY-----

homestar-runtime/fixtures/__testkey_ed25519_3.pem

@@ -0,0 +1,3 @@
+-----BEGIN PRIVATE KEY-----
+AgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgI=
+-----END PRIVATE KEY-----

homestar-runtime/src/event_handler.rs

@@ -58,6 +58,7 @@ pub(crate) struct EventHandler<DB: Database> {
     receiver: channel::AsyncBoundedChannelReceiver<Event>,
     query_senders: FnvHashMap<QueryId, (RequestResponseKey, Option<P2PSender>)>,
     connected_peers: FnvHashMap<PeerId, ConnectedPoint>,
+    connected_peers_limit: u32,
     request_response_senders: FnvHashMap<RequestId, (RequestResponseKey, P2PSender)>,
     rendezvous_cookies: FnvHashMap<PeerId, Cookie>,
     pubsub_enabled: bool,
@@ -80,6 +81,7 @@ pub(crate) struct EventHandler<DB: Database> {
     receiver: channel::AsyncBoundedChannelReceiver<Event>,
     query_senders: FnvHashMap<QueryId, (RequestResponseKey, Option<P2PSender>)>,
     connected_peers: FnvHashMap<PeerId, ConnectedPoint>,
+    connected_peers_limit: u32,
     request_response_senders: FnvHashMap<RequestId, (RequestResponseKey, P2PSender)>,
     rendezvous_cookies: FnvHashMap<PeerId, Cookie>,
     pubsub_enabled: bool,
@@ -121,6 +123,7 @@ where
             query_senders: FnvHashMap::default(),
             request_response_senders: FnvHashMap::default(),
             connected_peers: FnvHashMap::default(),
+            connected_peers_limit: settings.network.max_connected_peers,
             rendezvous_cookies: FnvHashMap::default(),
             pubsub_enabled: settings.network.enable_pubsub,
             ws_msg_sender,
@@ -144,6 +147,7 @@ where
             receiver,
             query_senders: FnvHashMap::default(),
             connected_peers: FnvHashMap::default(),
+            connected_peers_limit: settings.network.max_connected_peers,
             request_response_senders: FnvHashMap::default(),
             rendezvous_cookies: FnvHashMap::default(),
             pubsub_enabled: settings.network.enable_pubsub,

homestar-runtime/src/event_handler/swarm_event.rs

@@ -117,7 +117,7 @@ async fn handle_swarm_event<THandlerErr: fmt::Debug + Send, DB: Database>(
                     if num_addresses < event_handler.external_address_limit as usize {
                         info.observed_addr
                             .iter()
-                            // if _any_ part of the multiaddr includes a private IP, dont add it to our external address list
+                            // if _any_ part of the multiaddr includes a private IP, don't add it to our external address list
                             .filter_map(|proto| match proto {
                                 Protocol::Ip4(ip) => Some(ip),
                                 _ => None,
@@ -164,6 +164,7 @@ async fn handle_swarm_event<THandlerErr: fmt::Debug + Send, DB: Database>(
                                     "failed to register with rendezvous peer"
                                 )
                             }
+
                             // discover other nodes
                             rendezvous_client.discover(
                                 Some(Namespace::from_static(RENDEZVOUS_NAMESPACE)),
@@ -194,15 +195,25 @@ async fn handle_swarm_event<THandlerErr: fmt::Debug + Send, DB: Database>(
                             .insert(rendezvous_node, cookie);
 
                         // dial discovered peers
-                        for registration in registrations {
+                        for (index, registration) in registrations.iter().enumerate() {
                             // TODO: do anything with ttl here?
                             let opts = DialOpts::peer_id(registration.record.peer_id())
                                 .addresses(registration.record.addresses().to_vec())
                                 .condition(libp2p::swarm::dial_opts::PeerCondition::Disconnected)
                                 .build();
-                            // TODO: we might be dialing too many peers here. Add settings to configure when we stop dialing new peers
-                            if let Err(err) = event_handler.swarm.dial(opts) {
-                                warn!(peer_id=registration.record.peer_id().to_string(), err=?err, "failed to dial peer discovered through rendezvous")
+
+                            // Dial discovered peer if not at connected peers limit
+                            if event_handler.connected_peers.len() + index
+                                < event_handler.connected_peers_limit as usize
+                            {
+                                if let Err(err) = event_handler.swarm.dial(opts) {
+                                    warn!(peer_id=registration.record.peer_id().to_string(), err=?err, "failed to dial peer discovered through rendezvous")
+                                }
+                            } else {
+                                warn!(
+                                    peer_id=registration.record.peer_id().to_string(),
+                                    "peer discovered through rendezvous not dialed because max connected peers limit reached"
+                                )
                             }
                         }
                     } else {
@@ -261,13 +272,25 @@ async fn handle_swarm_event<THandlerErr: fmt::Debug + Send, DB: Database>(
                 rendezvous::server::Event::DiscoverNotServed { enquirer, error } => {
                     warn!(peer_id=enquirer.to_string(), err=?error, "did not serve rendezvous discover request")
                 }
+                rendezvous::server::Event::PeerRegistered { peer, .. } => {
+                    debug!(
+                        peer_id = peer.to_string(),
+                        "registered peer through rendezvous"
+                    )
+                }
                 rendezvous::server::Event::PeerNotRegistered {
                     peer,
                     namespace,
                     error,
                 } => {
                     warn!(peer_id=peer.to_string(), err=?error, namespace=?namespace, "did not register peer with rendezvous")
                 }
+                rendezvous::server::Event::RegistrationExpired(registration) => {
+                    debug!(
+                        peer_id = registration.record.peer_id().to_string(),
+                        "rendezvous peer registration expired on server"
+                    )
+                }
                 _ => (),
             }
         }
@@ -551,11 +574,21 @@ async fn handle_swarm_event<THandlerErr: fmt::Debug + Send, DB: Database>(
                     addr = multiaddr.to_string(),
                     "mDNS discovered a new peer"
                 );
-                let _ = event_handler.swarm.dial(
-                    DialOpts::peer_id(peer_id)
-                        .addresses(vec![multiaddr])
-                        .build(),
-                );
+
+                if event_handler.connected_peers.len()
+                    < event_handler.connected_peers_limit as usize
+                {
+                    let _ = event_handler.swarm.dial(
+                        DialOpts::peer_id(peer_id)
+                            .addresses(vec![multiaddr])
+                            .build(),
+                    );
+                } else {
+                    warn!(
+                        peer_id = peer_id.to_string(),
+                        "peer discovered by mDNS not dialed because max connected peers limit reached"
+                    )
+                }
             }
         }
         SwarmEvent::Behaviour(ComposedEvent::Mdns(mdns::Event::Expired(list))) => {

homestar-runtime/src/network/swarm.rs

@@ -98,12 +98,12 @@ pub(crate) async fn new(settings: &settings::Node) -> Result<Swarm<ComposedBehav
             } else {
                 None
             }),
-            rendezvous_client: Toggle::from(if settings.network.enable_rendezvous {
+            rendezvous_client: Toggle::from(if settings.network.enable_rendezvous_client {
                 Some(rendezvous::client::Behaviour::new(keypair.clone()))
             } else {
                 None
             }),
-            rendezvous_server: Toggle::from(if settings.network.enable_rendezvous {
+            rendezvous_server: Toggle::from(if settings.network.enable_rendezvous_server {
                 Some(rendezvous::server::Behaviour::new(
                     rendezvous::server::Config::default(),
                 ))
@@ -158,23 +158,27 @@ pub(crate) fn init(
     }
 
     // Dial nodes specified in settings. Failure here shouldn't halt node startup.
-    for addr in &settings.node_addresses {
-        let _ = swarm
-            .dial(addr.clone())
-            // log dial failure and continue
-            .map_err(|e| warn!(err=?e, "failed to dial configured node"));
-
-        // add node to kademlia routing table
-        if let Some(Protocol::P2p(peer_id)) =
-            addr.iter().find(|proto| matches!(proto, Protocol::P2p(_)))
-        {
-            info!(addr=?addr, "added configured node to kademlia routing table");
-            swarm
-                .behaviour_mut()
-                .kademlia
-                .add_address(&peer_id, addr.clone());
+    for (index, addr) in settings.node_addresses.iter().enumerate() {
+        if index < settings.max_connected_peers as usize {
+            let _ = swarm
+                .dial(addr.clone())
+                // log dial failure and continue
+                .map_err(|e| warn!(err=?e, "failed to dial configured node"));
+
+            // add node to kademlia routing table
+            if let Some(Protocol::P2p(peer_id)) =
+                addr.iter().find(|proto| matches!(proto, Protocol::P2p(_)))
+            {
+                info!(addr=?addr, "added configured node to kademlia routing table");
+                swarm
+                    .behaviour_mut()
+                    .kademlia
+                    .add_address(&peer_id, addr.clone());
+            } else {
+                warn!(addr=?addr, err="configured node address did not include a peer ID", "node not added to kademlia routing table")
+            }
         } else {
-            warn!(addr=?addr, err="configured node address did not include a peer ID", "node not added to kademlia routing table")
+            warn!(addr=?addr, "address not dialed because node addresses count exceeds max connected peers configuration")
         }
     }
 

homestar-runtime/src/settings.rs

@@ -76,8 +76,10 @@ pub struct Network {
     /// [Swarm]: libp2p::swarm::Swarm
     #[serde(with = "http_serde::uri")]
     pub(crate) listen_address: Uri,
-    /// Enable Rendezvous protocol.
-    pub(crate) enable_rendezvous: bool,
+    /// Enable Rendezvous protocol client.
+    pub(crate) enable_rendezvous_client: bool,
+    /// Enable Rendezvous protocol server.
+    pub(crate) enable_rendezvous_server: bool,
     /// Enable mDNS.
     pub(crate) enable_mdns: bool,
     /// mDNS IPv6 enable flag
@@ -141,7 +143,9 @@ pub struct Network {
     /// network.
     #[serde_as(as = "Vec<serde_with::DisplayFromStr>")]
     pub(crate) announce_addresses: Vec<libp2p::Multiaddr>,
-    /// Limit on the number of external addresses we annoucne to other peers.
+    /// Maximum number of peers we will dial.
+    pub(crate) max_connected_peers: u32,
+    /// Limit on the number of external addresses we announce to other peers.
     pub(crate) max_announce_addresses: u32,
 }
 
@@ -186,8 +190,9 @@ impl Default for Network {
         Self {
             events_buffer_len: 1024,
             listen_address: Uri::from_static("/ip4/0.0.0.0/tcp/0"),
+            enable_rendezvous_client: true,
+            enable_rendezvous_server: false,
             // TODO: we would like to enable this by default, however this breaks mdns on at least some linux distros. Requires further investigation.
-            enable_rendezvous: true,
             enable_mdns: true,
             mdns_enable_ipv6: false,
             mdns_query_interval: Duration::from_secs(5 * 60),
@@ -212,6 +217,7 @@ impl Default for Network {
             node_addresses: Vec::new(),
             announce_addresses: Vec::new(),
             max_announce_addresses: 10,
+            max_connected_peers: 32,
         }
     }
 }

homestar-runtime/tests/fixtures/test_mdns1.toml

@@ -9,7 +9,7 @@ console_subscriber_port = 5560
 rpc_port = 9800
 websocket_port = 8000
 listen_address = "/ip4/0.0.0.0/tcp/0"
-enable_rendezvous = false
+enable_rendezvous_client = false
 
 [node.network.keypair_config]
 existing = { key_type = "ed25519", path = "./fixtures/__testkey_ed25519.pem" }

homestar-runtime/tests/fixtures/test_mdns2.toml

@@ -9,7 +9,7 @@ console_subscriber_port = 5561
 rpc_port = 9801
 websocket_port = 8001
 listen_address = "/ip4/0.0.0.0/tcp/0"
-enable_rendezvous = false
+enable_rendezvous_client = false
 
 [node.network.keypair_config]
 existing = { key_type = "secp256k1", path = "./fixtures/__testkey_secp256k1.der" }

homestar-runtime/tests/fixtures/test_network1.toml

@@ -13,7 +13,7 @@ node_addresses = [
   "/ip4/127.0.0.1/tcp/7001/p2p/16Uiu2HAm3g9AomQNeEctL2hPwLapap7AtPSNt8ZrBny4rLx1W5Dc",
 ]
 enable_mdns = false
-enable_rendezvous = false
+enable_rendezvous_client = false
 
 [node.network.keypair_config]
 existing = { key_type = "ed25519", path = "./fixtures/__testkey_ed25519.pem" }

homestar-runtime/tests/fixtures/test_network2.toml

@@ -13,7 +13,7 @@ node_addresses = [
   "/ip4/127.0.0.1/tcp/7000/p2p/12D3KooWDpJ7As7BWAwRMfu1VU2WCqNjvq387JEYKDBj4kx6nXTN",
 ]
 enable_mdns = false
-enable_rendezvous = false
+enable_rendezvous_client = false
 
 [node.network.keypair_config]
 existing = { key_type = "secp256k1", path = "./fixtures/__testkey_secp256k1.der" }

homestar-runtime/tests/fixtures/test_rendezvous1.toml

@@ -0,0 +1,16 @@
+[monitoring]
+process_collector_interval = 500
+metrics_port = 4035
+console_subscriber_port = 5585
+
+[node]
+
+[node.network]
+rpc_port = 9825
+websocket_port = 8025
+listen_address = "/ip4/127.0.0.1/tcp/7000"
+enable_rendezvous_server = true
+enable_mdns = false
+
+[node.network.keypair_config]
+existing = { key_type = "ed25519", path = "./fixtures/__testkey_ed25519.pem" }

homestar-runtime/tests/fixtures/test_rendezvous2.toml

@@ -0,0 +1,21 @@
+[monitoring]
+process_collector_interval = 500
+metrics_port = 4036
+console_subscriber_port = 5586
+
+[node]
+
+[node.network]
+rpc_port = 9826
+websocket_port = 8026
+listen_address = "/ip4/127.0.0.1/tcp/7001"
+announce_addresses = [
+  "/ip4/127.0.0.1/tcp/7001/p2p/16Uiu2HAm3g9AomQNeEctL2hPwLapap7AtPSNt8ZrBny4rLx1W5Dc",
+]
+node_addresses = [
+  "/ip4/127.0.0.1/tcp/7000/p2p/12D3KooWDpJ7As7BWAwRMfu1VU2WCqNjvq387JEYKDBj4kx6nXTN",
+]
+enable_mdns = false
+
+[node.network.keypair_config]
+existing = { key_type = "secp256k1", path = "./fixtures/__testkey_secp256k1.der" }

homestar-runtime/tests/fixtures/test_rendezvous3.toml

@@ -0,0 +1,18 @@
+[monitoring]
+process_collector_interval = 500
+metrics_port = 4037
+console_subscriber_port = 5587
+
+[node]
+
+[node.network]
+rpc_port = 9827
+websocket_port = 8027
+listen_address = "/ip4/127.0.0.1/tcp/7002"
+node_addresses = [
+  "/ip4/127.0.0.1/tcp/7000/p2p/12D3KooWDpJ7As7BWAwRMfu1VU2WCqNjvq387JEYKDBj4kx6nXTN",
+]
+enable_mdns = false
+
+[node.network.keypair_config]
+existing = { key_type = "ed25519", path = "./fixtures/__testkey_ed25519_2.pem" }