Support client mode in kademlia

.gitignore

@@ -1,2 +1,3 @@
 target
 Cargo.lock
+.vscode/launch.json

protocols/kad/src/behaviour.rs

@@ -29,7 +29,7 @@ use crate::handler::{
 };
 use crate::jobs::*;
 use crate::kbucket::{self, Distance, KBucketsTable, NodeStatus};
-use crate::protocol::{KadConnectionType, KadPeer, KademliaProtocolConfig};
+use crate::protocol::{KadConnectionType, KadPeer, KademliaProtocolConfig, Mode};
 use crate::query::{Query, QueryConfig, QueryId, QueryPool, QueryPoolState};
 use crate::record::{
     self,
@@ -175,6 +175,8 @@ pub struct KademliaConfig {
     connection_idle_timeout: Duration,
     kbucket_inserts: KademliaBucketInserts,
     caching: KademliaCaching,
+    // Set the peer mode.
+    mode: Mode,
 }
 
 /// The configuration for Kademlia "write-back" caching after successful
@@ -209,6 +211,8 @@ impl Default for KademliaConfig {
             connection_idle_timeout: Duration::from_secs(10),
             kbucket_inserts: KademliaBucketInserts::OnConnected,
             caching: KademliaCaching::Enabled { max_peers: 1 },
+            // By default, a peer will not be in client mode.
+            mode: Mode::default(),
         }
     }
 }
@@ -390,6 +394,21 @@ impl KademliaConfig {
         self.caching = c;
         self
     }
+
+    /// Sets the [`Mode`] the node is operating in.
+    ///
+    /// The default is [`Mode::Server`].
+    ///
+    /// In [`Mode::Client`] the node does not advertise support for the Kademlia
+    /// protocol, nor does it accept incoming Kademlia streams. Peers will thus not
+    /// include the local node in their routing table. The node can still make
+    /// outbound requests.
+    ///
+    /// Use [`Mode::Client`] when not publicly reachable.
+    pub fn set_mode(&mut self, mode: Mode) -> &mut Self {
+        self.mode = mode;
+        self
+    }
 }
 
 impl<TStore> Kademlia<TStore>
@@ -1710,6 +1729,7 @@ where
             protocol_config: self.protocol_config.clone(),
             allow_listening: true,
             idle_timeout: self.connection_idle_timeout,
+            mode: Mode::default(),
         })
     }
 

protocols/kad/src/behaviour/test.rs

@@ -1317,3 +1317,96 @@ fn network_behaviour_inject_address_change() {
         kademlia.addresses_of_peer(&remote_peer_id),
     );
 }
+
+#[test]
+fn client_mode() {
+    // Create server peers.
+    let mut server_nodes = build_fully_connected_nodes_with_config(2, KademliaConfig::default());
+
+    // Create a client peer.
+    let mut cfg = KademliaConfig::default();
+    cfg.set_mode(Mode::Client);
+    let mut client = build_node_with_config(cfg);
+
+    // Fitler out peer Ids.
+    let mut peers: Vec<_> = server_nodes
+        .iter()
+        .map(|(_, swarm)| swarm.local_peer_id().clone())
+        .collect();
+    peers.push(client.1.local_peer_id().clone());
+
+    // Filter out MultiAddrs and swarms.
+    let mut addrs: Vec<_> = server_nodes.iter().map(|(addr, _)| addr.clone()).collect();
+    addrs.push(client.0.clone());
+
+    let mut swarms: Vec<_> = server_nodes.iter_mut().map(|(_, swarm)| swarm).collect();
+    swarms.push(&mut client.1);
+
+    let key = "123".to_string();
+    let key = Key::new(&key);
+    let record = Record::new(key.clone(), vec![4, 5, 6]);
+
+    // Put a record from a server peer.
+    swarms[0]
+        .behaviour_mut()
+        .put_record(record.clone(), Quorum::One)
+        .unwrap();
+
+    // Dial a server peer from the client peer.
+    swarms[2]
+        .behaviour_mut()
+        .add_address(&peers[1], addrs[1].clone());
+
+    // Try to get the same record from the client node.
+    swarms[2].behaviour_mut().get_record(&key, Quorum::One);
+
+    block_on(poll_fn(|ctx| {
+        for swarm in &mut swarms {
+            loop {
+                match swarm.poll_next_unpin(ctx) {
+                    Poll::Ready(Some(SwarmEvent::Behaviour(
+                        KademliaEvent::OutboundQueryCompleted { result, .. },
+                    ))) => match result {
+                        QueryResult::PutRecord(result) => match result {
+                            Ok(PutRecordOk { .. }) => {
+                                // Check if the server peer is not connected to the client peer.
+                                assert!(
+                                    swarm
+                                        .behaviour_mut()
+                                        .connected_peers
+                                        .iter()
+                                        .all(|p| *p != peers[2]),
+                                    "The server peer is connected to the client peer."
+                                );
+                            }
+                            Err(e) => panic!("PUT operation failed: {:?}", e),
+                        },
+                        QueryResult::GetRecord(result) => match result {
+                            Ok(GetRecordOk { .. }) => {
+                                // Check if the client peer is connected to the server peer.
+                                assert!(
+                                    swarm
+                                        .behaviour_mut()
+                                        .connected_peers
+                                        .iter()
+                                        .any(|p| (*p == peers[0] || *p == peers[1])),
+                                    "The client peer is not connected to the server peers."
+                                );
+
+                                // GetRecord will never be successful unless and until PutRecord is
+                                // successful.
+                                return Poll::Ready(());
+                            }
+                            Err(e) => panic!("GET operation failed: {:?}", e),
+                        },
+                        _ => {}
+                    },
+                    Poll::Ready(Some(_)) => {}
+                    e @ Poll::Ready(_) => panic!("Unexpected return value: {:?}", e),
+                    Poll::Pending => break,
+                }
+            }
+        }
+        Poll::Pending
+    }));
+}

protocols/kad/src/handler.rs

@@ -20,7 +20,7 @@
 
 use crate::protocol::{
     KadInStreamSink, KadOutStreamSink, KadPeer, KadRequestMsg, KadResponseMsg,
-    KademliaProtocolConfig,
+    KademliaProtocolConfig, Mode,
 };
 use crate::record::{self, Record};
 use futures::prelude::*;
@@ -62,10 +62,15 @@ impl<T: Clone + fmt::Debug + Send + 'static> IntoProtocolsHandler for KademliaHa
     }
 
     fn inbound_protocol(&self) -> <Self::Handler as ProtocolsHandler>::InboundProtocol {
-        if self.config.allow_listening {
-            upgrade::EitherUpgrade::A(self.config.protocol_config.clone())
-        } else {
-            upgrade::EitherUpgrade::B(upgrade::DeniedUpgrade)
+        match self.config.mode {
+            Mode::Client => upgrade::EitherUpgrade::B(upgrade::DeniedUpgrade),
+            Mode::Server => {
+                if self.config.allow_listening {
+                    upgrade::EitherUpgrade::A(self.config.protocol_config.clone())
+                } else {
+                    upgrade::EitherUpgrade::B(upgrade::DeniedUpgrade)
+                }
+            }
         }
     }
 }
@@ -123,6 +128,9 @@ pub struct KademliaHandlerConfig {
 
     /// Time after which we close an idle connection.
     pub idle_timeout: Duration,
+
+    /// [`Mode`] the handler is operating in.
+    pub mode: Mode,
 }
 
 /// State of an active substream, opened either by us or by the remote.
@@ -480,11 +488,20 @@ where
     type InboundOpenInfo = ();
 
     fn listen_protocol(&self) -> SubstreamProtocol<Self::InboundProtocol, Self::InboundOpenInfo> {
-        if self.config.allow_listening {
-            SubstreamProtocol::new(self.config.protocol_config.clone(), ())
-                .map_upgrade(upgrade::EitherUpgrade::A)
-        } else {
-            SubstreamProtocol::new(upgrade::EitherUpgrade::B(upgrade::DeniedUpgrade), ())
+        match self.config.mode {
+            Mode::Server => {
+                if self.config.allow_listening {
+                    SubstreamProtocol::new(self.config.protocol_config.clone(), ())
+                        .map_upgrade(upgrade::EitherUpgrade::A)
+                } else {
+                    SubstreamProtocol::new(upgrade::EitherUpgrade::B(upgrade::DeniedUpgrade), ())
+                }
+            }
+            // If we are in client mode, I don't want to advertise Kademlia so that other peers will
+            // not send me Kademlia requests.
+            Mode::Client => {
+                SubstreamProtocol::new(upgrade::EitherUpgrade::B(upgrade::DeniedUpgrade), ())
+            }
         }
     }
 
@@ -520,12 +537,8 @@ where
         self.next_connec_unique_id.0 += 1;
         self.substreams
             .push(SubstreamState::InWaitingMessage(connec_unique_id, protocol));
-        if let ProtocolStatus::Unconfirmed = self.protocol_status {
-            // Upon the first successfully negotiated substream, we know that the
-            // remote is configured with the same protocol name and we want
-            // the behaviour to add this peer to the routing table, if possible.
-            self.protocol_status = ProtocolStatus::Confirmed;
-        }
+        // Just because another peer is sending us Kademlia requests, doesn't necessarily
+        // mean that it will answer the Kademlia requests that we send to it.
     }
 
     fn inject_event(&mut self, message: KademliaHandlerIn<TUserData>) {
@@ -763,6 +776,7 @@ impl Default for KademliaHandlerConfig {
             protocol_config: Default::default(),
             allow_listening: true,
             idle_timeout: Duration::from_secs(10),
+            mode: Mode::default(),
         }
     }
 }

protocols/kad/src/protocol.rs

@@ -46,6 +46,19 @@ pub const DEFAULT_PROTO_NAME: &[u8] = b"/ipfs/kad/1.0.0";
 /// The default maximum size for a varint length-delimited packet.
 pub const DEFAULT_MAX_PACKET_SIZE: usize = 16 * 1024;
 
+#[derive(Debug, Clone)]
+/// See [`crate::KademliaConfig::set_mode`].
+pub enum Mode {
+    Client,
+    Server,
+}
+
+impl Default for Mode {
+    fn default() -> Self {
+        Mode::Server
+    }
+}
+
 /// Status of our connection to a node reported by the Kademlia protocol.
 #[derive(Copy, Clone, PartialEq, Eq, Debug, Hash)]
 pub enum KadConnectionType {