Swarm Bootstrap Nodes

homestar-runtime/fixtures/settings.toml

@@ -6,3 +6,4 @@ process_collector_interval = 10
 [node.network]
 events_buffer_len = 1000
 websocket_port = 9999
+trusted_node_addresses = ["/ip4/127.0.0.1/tcp/9998/ws"]

homestar-runtime/src/event_handler.rs

@@ -10,7 +10,10 @@ use crate::{
 use anyhow::Result;
 use async_trait::async_trait;
 use fnv::FnvHashMap;
-use libp2p::{futures::StreamExt, kad::QueryId, request_response::RequestId, swarm::Swarm};
+use libp2p::{
+    core::ConnectedPoint, futures::StreamExt, kad::QueryId, request_response::RequestId,
+    swarm::Swarm, PeerId,
+};
 use std::{sync::Arc, time::Duration};
 use swarm_event::ResponseEvent;
 use tokio::{select, sync::mpsc};
@@ -48,6 +51,7 @@ pub(crate) struct EventHandler<DB: Database> {
     sender: Arc<mpsc::Sender<Event>>,
     receiver: mpsc::Receiver<Event>,
     query_senders: FnvHashMap<QueryId, (RequestResponseKey, P2PSender)>,
+    connected_peers: FnvHashMap<PeerId, ConnectedPoint>,
     request_response_senders: FnvHashMap<RequestId, (RequestResponseKey, P2PSender)>,
 }
 
@@ -71,6 +75,7 @@ where
             sender: Arc::new(sender),
             receiver,
             query_senders: FnvHashMap::default(),
+            connected_peers: FnvHashMap::default(),
             request_response_senders: FnvHashMap::default(),
         }
     }

homestar-runtime/src/event_handler/swarm_event.rs

@@ -376,6 +376,26 @@ async fn handle_swarm_event<THandlerErr: fmt::Debug + Send, DB: Database>(
             );
         }
         SwarmEvent::IncomingConnection { .. } => {}
+        SwarmEvent::ConnectionEstablished {
+            peer_id, endpoint, ..
+        } => {
+            let behavior = event_handler.swarm.behaviour_mut();
+            // only listener addresses should be added to the routing table.
+            // TODO: add identify here to discover more listen addresses and add those.
+            if endpoint.is_listener() {
+                // ignores if the peer failed or not to be added in the routing table.
+                behavior
+                    .kademlia
+                    .add_address(&peer_id, endpoint.get_remote_address().clone());
+            }
+
+            // add peer to connected peers list
+            event_handler.connected_peers.insert(peer_id, endpoint);
+        }
+        SwarmEvent::ConnectionClosed { peer_id, cause, .. } => {
+            info!("peer connection closed {peer_id}, cause: {cause:?}");
+            event_handler.connected_peers.remove_entry(&peer_id);
+        }
         _ => {}
     }
 }

homestar-runtime/src/network/swarm.rs

@@ -18,6 +18,7 @@ use libp2p::{
 };
 use serde::{Deserialize, Serialize};
 use std::fmt;
+use tracing::{info, warn};
 
 /// Build a new [Swarm] with a given transport and a tokio executor.
 pub(crate) async fn new(settings: &settings::Node) -> Result<Swarm<ComposedBehaviour>> {
@@ -54,8 +55,7 @@ pub(crate) async fn new(settings: &settings::Node) -> Result<Swarm<ComposedBehav
     )
     .build();
 
-    // Listen-on given address
-    swarm.listen_on(settings.network.listen_address.to_string().parse()?)?;
+    startup(&mut swarm, &settings.network)?;
 
     // subscribe to `receipts` topic
     swarm
@@ -65,6 +65,24 @@ pub(crate) async fn new(settings: &settings::Node) -> Result<Swarm<ComposedBehav
     Ok(swarm)
 }
 
+fn startup<T: NetworkBehaviour>(swarm: &mut Swarm<T>, settings: &settings::Network) -> Result<()> {
+    // Listen-on given address
+    swarm.listen_on(settings.listen_address.to_string().parse()?)?;
+
+    // Dial trusted nodes specified in settings. Failure here shouldn't halt node startup.
+    for trusted_addr in &settings.trusted_node_addresses {
+        swarm
+            .dial(trusted_addr.clone())
+            .map(|_| {
+                info!(trusted_address=?trusted_addr, "Successfully dialed configured trusted node")
+            })
+            // log dial failure and continue
+            .map_err(|e| warn!(err=?e, "Failed to dial trusted node"))
+            .ok();
+    }
+    Ok(())
+}
+
 /// Key data structure for [request_response::Event] messages.
 #[derive(Debug, Clone, Serialize, Deserialize, Encode, Decode)]
 pub(crate) struct RequestResponseKey {

homestar-runtime/src/settings.rs

@@ -98,6 +98,9 @@ pub struct Network {
     pub(crate) workflow_quorum: usize,
     /// Pubkey setup configuration
     pub(crate) keypair_config: PubkeyConfig,
+    /// Multiaddrs of the trusted nodes to connect to on startup. These addresses are added as explicit peers for gossipsub.
+    #[serde_as(as = "Vec<serde_with::DisplayFromStr>")]
+    pub(crate) trusted_node_addresses: Vec<libp2p::Multiaddr>,
 }
 
 /// Database-related settings for a homestar node.
@@ -173,6 +176,7 @@ impl Default for Network {
             websocket_capacity: 100,
             workflow_quorum: 3,
             keypair_config: PubkeyConfig::Random,
+            trusted_node_addresses: Vec::new(),
         }
     }
 }
@@ -323,6 +327,8 @@ mod test {
         default_modded_settings.network.events_buffer_len = 1000;
         default_modded_settings.network.websocket_port = 9999;
         default_modded_settings.shutdown_timeout = Duration::from_secs(20);
+        default_modded_settings.network.trusted_node_addresses =
+            vec!["/ip4/127.0.0.1/tcp/9998/ws".to_string().try_into().unwrap()];
         assert_eq!(settings.node(), &default_modded_settings);
     }