kad: Allow connecting to more than one DHT network (#473)

Needed for IPFS support in substrate, because we need to connect to
polkadot & IPFS DHTs simultaneously.

Close #471.

Author: dmitry-markin

src/config.rs

@@ -93,7 +93,7 @@ pub struct ConfigBuilder {
     identify: Option<identify::Config>,
 
     /// Kademlia protocol config.
-    kademlia: Option<kademlia::Config>,
+    kademlia: Vec<kademlia::Config>,
 
     /// Bitswap protocol config.
     bitswap: Option<bitswap::Config>,
@@ -149,7 +149,7 @@ impl ConfigBuilder {
             keypair: None,
             ping: None,
             identify: None,
-            kademlia: None,
+            kademlia: Vec::new(),
             bitswap: None,
             mdns: None,
             executor: None,
@@ -219,7 +219,7 @@ impl ConfigBuilder {
 
     /// Enable IPFS Kademlia protocol.
     pub fn with_libp2p_kademlia(mut self, config: kademlia::Config) -> Self {
-        self.kademlia = Some(config);
+        self.kademlia.push(config);
         self
     }
 
@@ -307,7 +307,7 @@ impl ConfigBuilder {
             websocket: self.websocket.take(),
             ping: self.ping.take(),
             identify: self.identify.take(),
-            kademlia: self.kademlia.take(),
+            kademlia: self.kademlia,
             bitswap: self.bitswap.take(),
             max_parallel_dials: self.max_parallel_dials,
             executor: self.executor.map_or(Arc::new(DefaultExecutor {}), |executor| executor),
@@ -349,7 +349,7 @@ pub struct Litep2pConfig {
     pub(crate) identify: Option<identify::Config>,
 
     /// Kademlia protocol configuration, if enabled.
-    pub(crate) kademlia: Option<kademlia::Config>,
+    pub(crate) kademlia: Vec<kademlia::Config>,
 
     /// Bitswap protocol configuration, if enabled.
     pub(crate) bitswap: Option<bitswap::Config>,

src/lib.rs

@@ -259,8 +259,8 @@ impl Litep2p {
             }));
         }
 
-        // start kademlia protocol event loop if enabled
-        if let Some(kademlia_config) = litep2p_config.kademlia.take() {
+        // start kademlia protocol event loops
+        for kademlia_config in litep2p_config.kademlia.into_iter() {
             tracing::debug!(
                 target: LOG_TARGET,
                 protocol_names = ?kademlia_config.protocol_names,

src/protocol/libp2p/kademlia/mod.rs

@@ -240,7 +240,9 @@ impl Kademlia {
                 self.routing_table.on_connection_established(Key::from(peer), endpoint);
 
                 let Some(actions) = self.pending_dials.remove(&peer) else {
-                    entry.insert(PeerContext::new());
+                    // Note that we do not add peer entry if we don't have any pending actions.
+                    // This is done to not populate `self.peers` with peers that don't support
+                    // our Kademlia protocol.
                     return Ok(());
                 };
 
@@ -343,6 +345,7 @@ impl Kademlia {
         let pending_action = &mut self
             .peers
             .get_mut(&peer)
+            // If we opened an outbound substream, we must have pending actions for the peer.
             .ok_or(Error::PeerDoesntExist(peer))?
             .pending_actions
             .remove(&substream_id);
@@ -412,6 +415,10 @@ impl Kademlia {
     async fn on_inbound_substream(&mut self, peer: PeerId, substream: Substream) {
         tracing::trace!(target: LOG_TARGET, ?peer, "inbound substream opened");
 
+        // Ensure peer entry exists to treat peer as [`ConnectionType::Connected`].
+        // when inserting into the routing table.
+        self.peers.entry(peer).or_default();
+
         self.executor.read_message(peer, None, substream);
     }