feat: report changes in supported protocols to ConnectionHandler

protocols/dcutr/src/handler/direct.rs

@@ -92,7 +92,9 @@ impl ConnectionHandler for Handler {
             | ConnectionEvent::FullyNegotiatedOutbound(_)
             | ConnectionEvent::DialUpgradeError(_)
             | ConnectionEvent::ListenUpgradeError(_)
-            | ConnectionEvent::AddressChange(_) => {}
+            | ConnectionEvent::AddressChange(_)
+            | ConnectionEvent::LocalProtocolsChange(_)
+            | ConnectionEvent::RemoteProtocolsChange(_) => {}
         }
     }
 }

protocols/dcutr/src/handler/relayed.rs

@@ -419,7 +419,9 @@ impl ConnectionHandler for Handler {
             ConnectionEvent::DialUpgradeError(dial_upgrade_error) => {
                 self.on_dial_upgrade_error(dial_upgrade_error)
             }
-            ConnectionEvent::AddressChange(_) => {}
+            ConnectionEvent::AddressChange(_)
+            | ConnectionEvent::LocalProtocolsChange(_)
+            | ConnectionEvent::RemoteProtocolsChange(_) => {}
         }
     }
 }

protocols/gossipsub/src/handler.rs

@@ -576,7 +576,10 @@ impl ConnectionHandler for Handler {
                 warn!("Dial upgrade error {:?}", e);
                 self.upgrade_errors.push_back(e);
             }
-            ConnectionEvent::AddressChange(_) | ConnectionEvent::ListenUpgradeError(_) => {}
+            ConnectionEvent::AddressChange(_)
+            | ConnectionEvent::ListenUpgradeError(_)
+            | ConnectionEvent::LocalProtocolsChange(_)
+            | ConnectionEvent::RemoteProtocolsChange(_) => {}
         }
     }
 }

protocols/identify/src/behaviour.rs

@@ -323,7 +323,7 @@ impl NetworkBehaviour for Behaviour {
     fn poll(
         &mut self,
         _cx: &mut Context<'_>,
-        params: &mut impl PollParameters,
+        _: &mut impl PollParameters,
     ) -> Poll<ToSwarm<Self::OutEvent, THandlerInEvent<Self>>> {
         if let Some(event) = self.events.pop_front() {
             return Poll::Ready(event);
@@ -344,7 +344,6 @@ impl NetworkBehaviour for Behaviour {
                         .chain(self.external_addresses.iter())
                         .cloned()
                         .collect(),
-                    supported_protocols: supported_protocols(params),
                     protocol: Protocol::Push,
                 },
             }),
@@ -361,7 +360,6 @@ impl NetworkBehaviour for Behaviour {
                         .chain(self.external_addresses.iter())
                         .cloned()
                         .collect(),
-                    supported_protocols: supported_protocols(params),
                     protocol: Protocol::Identify(connection_id),
                 },
             }),
@@ -488,15 +486,6 @@ pub enum Event {
     },
 }
 
-fn supported_protocols(params: &impl PollParameters) -> Vec<String> {
-    // The protocol names can be bytes, but the identify protocol except UTF-8 strings.
-    // There's not much we can do to solve this conflict except strip non-UTF-8 characters.
-    params
-        .supported_protocols()
-        .map(|p| String::from_utf8_lossy(&p).to_string())
-        .collect()
-}
-
 /// If there is a given peer_id in the multiaddr, make sure it is the same as
 /// the given peer_id. If there is no peer_id for the peer in the mutiaddr, this returns true.
 fn multiaddr_matches_peer_id(addr: &Multiaddr, peer_id: &PeerId) -> bool {

protocols/identify/src/handler.rs

@@ -32,6 +32,7 @@ use libp2p_identity::PeerId;
 use libp2p_identity::PublicKey;
 use libp2p_swarm::handler::{
     ConnectionEvent, DialUpgradeError, FullyNegotiatedInbound, FullyNegotiatedOutbound,
+    ProtocolsChange,
 };
 use libp2p_swarm::{
     ConnectionHandler, ConnectionHandlerEvent, ConnectionHandlerUpgrErr, KeepAlive,
@@ -83,6 +84,8 @@ pub struct Handler {
 
     /// Address observed by or for the remote.
     observed_addr: Multiaddr,
+
+    local_supported_protocols: Vec<String>,
 }
 
 /// An event from `Behaviour` with the information requested by the `Handler`.
@@ -91,9 +94,6 @@ pub struct InEvent {
     /// The addresses that the peer is listening on.
     pub listen_addrs: Vec<Multiaddr>,
 
-    /// The list of protocols supported by the peer, e.g. `/ipfs/ping/1.0.0`.
-    pub supported_protocols: Vec<String>,
-
     /// The protocol w.r.t. the information requested.
     pub protocol: Protocol,
 }
@@ -138,6 +138,7 @@ impl Handler {
             protocol_version,
             agent_version,
             observed_addr,
+            local_supported_protocols: vec![],
         }
     }
 
@@ -186,6 +187,10 @@ impl Handler {
     ) {
         match output {
             future::Either::Left(remote_info) => {
+                self.events
+                    .push(ConnectionHandlerEvent::ReportRemoteProtocols {
+                        protocols: remote_info.protocols.clone(),
+                    });
                 self.events
                     .push(ConnectionHandlerEvent::Custom(Event::Identified(
                         remote_info,
@@ -238,7 +243,6 @@ impl ConnectionHandler for Handler {
         &mut self,
         InEvent {
             listen_addrs,
-            supported_protocols,
             protocol,
         }: Self::InEvent,
     ) {
@@ -247,7 +251,7 @@ impl ConnectionHandler for Handler {
             protocol_version: self.protocol_version.clone(),
             agent_version: self.agent_version.clone(),
             listen_addrs,
-            protocols: supported_protocols,
+            protocols: self.local_supported_protocols.clone(),
             observed_addr: self.observed_addr.clone(),
         };
 
@@ -307,6 +311,10 @@ impl ConnectionHandler for Handler {
             self.inbound_identify_push.take();
 
             if let Ok(info) = res {
+                self.events
+                    .push(ConnectionHandlerEvent::ReportRemoteProtocols {
+                        protocols: info.protocols.clone(),
+                    });
                 return Poll::Ready(ConnectionHandlerEvent::Custom(Event::Identified(info)));
             }
         }
@@ -345,6 +353,10 @@ impl ConnectionHandler for Handler {
                 self.on_dial_upgrade_error(dial_upgrade_error)
             }
             ConnectionEvent::AddressChange(_) | ConnectionEvent::ListenUpgradeError(_) => {}
+            ConnectionEvent::LocalProtocolsChange(ProtocolsChange { protocols }) => {
+                self.local_supported_protocols = protocols.to_vec();
+            }
+            ConnectionEvent::RemoteProtocolsChange(_) => {}
         }
     }
 }

protocols/kad/Cargo.toml

@@ -34,9 +34,13 @@ serde = { version = "1.0", optional = true, features = ["derive"] }
 thiserror = "1"
 
 [dev-dependencies]
+async-std = { version = "1.12.0", features = ["attributes"] }
 env_logger = "0.10.0"
 futures-timer = "3.0"
+libp2p-identify = { path = "../identify" }
 libp2p-noise = { path = "../../transports/noise" }
+libp2p-swarm = { path = "../../swarm", features = ["macros"] }
+libp2p-swarm-test = { path = "../../swarm-test" }
 libp2p-yamux = { path = "../../muxers/yamux" }
 quickcheck = { package = "quickcheck-ext", path = "../../misc/quickcheck-ext" }
 

protocols/kad/src/behaviour.rs

@@ -116,6 +116,7 @@ pub struct Kademlia<TStore> {
 
     /// The record storage.
     store: TStore,
+    mode: Mode,
 }
 
 /// The configurable strategies for the insertion of peers
@@ -182,6 +183,7 @@ pub struct KademliaConfig {
     connection_idle_timeout: Duration,
     kbucket_inserts: KademliaBucketInserts,
     caching: KademliaCaching,
+    mode: Mode,
 }
 
 impl Default for KademliaConfig {
@@ -199,6 +201,7 @@ impl Default for KademliaConfig {
             connection_idle_timeout: Duration::from_secs(10),
             kbucket_inserts: KademliaBucketInserts::OnConnected,
             caching: KademliaCaching::Enabled { max_peers: 1 },
+            mode: Mode::Server,
         }
     }
 }
@@ -399,6 +402,14 @@ impl KademliaConfig {
         self.caching = c;
         self
     }
+
+    /// Sets the mode.
+    ///
+    /// TODO: More docs.
+    pub fn set_mode(&mut self, m: Mode) -> &mut Self {
+        self.mode = m;
+        self
+    }
 }
 
 impl<TStore> Kademlia<TStore>
@@ -453,6 +464,7 @@ where
             connection_idle_timeout: config.connection_idle_timeout,
             external_addresses: Default::default(),
             local_peer_id: id,
+            mode: config.mode,
         }
     }
 
@@ -1976,7 +1988,7 @@ where
         Ok(KademliaHandler::new(
             KademliaHandlerConfig {
                 protocol_config: self.protocol_config.clone(),
-                allow_listening: true,
+                allow_listening: self.mode == Mode::Server,
                 idle_timeout: self.connection_idle_timeout,
             },
             ConnectedPoint::Listener {
@@ -1997,7 +2009,7 @@ where
         Ok(KademliaHandler::new(
             KademliaHandlerConfig {
                 protocol_config: self.protocol_config.clone(),
-                allow_listening: true,
+                allow_listening: self.mode == Mode::Server,
                 idle_timeout: self.connection_idle_timeout,
             },
             ConnectedPoint::Dialer {
@@ -2062,6 +2074,14 @@ where
                 self.connection_updated(source, address, NodeStatus::Connected);
             }
 
+            KademliaHandlerEvent::ProtocolNotSupported { endpoint } => {
+                let address = match endpoint {
+                    ConnectedPoint::Dialer { address, .. } => Some(address),
+                    ConnectedPoint::Listener { .. } => None,
+                };
+                self.connection_updated(source, address, NodeStatus::Disconnected);
+            }
+
             KademliaHandlerEvent::FindNodeReq { key, request_id } => {
                 let closer_peers = self.find_closest(&kbucket::Key::new(key), &source);
 
@@ -3190,3 +3210,9 @@ pub enum RoutingUpdate {
     /// peer ID).
     Failed,
 }
+
+#[derive(Debug, Clone, Copy, PartialEq)]
+pub enum Mode {
+    Client,
+    Server,
+}

protocols/kad/src/handler.rs

@@ -31,6 +31,7 @@ use libp2p_core::{upgrade, ConnectedPoint};
 use libp2p_identity::PeerId;
 use libp2p_swarm::handler::{
     ConnectionEvent, DialUpgradeError, FullyNegotiatedInbound, FullyNegotiatedOutbound,
+    ProtocolsChange,
 };
 use libp2p_swarm::{
     ConnectionHandler, ConnectionHandlerEvent, ConnectionHandlerUpgrErr, KeepAlive,
@@ -92,10 +93,12 @@ pub struct KademliaHandler<TUserData> {
 enum ProtocolStatus {
     /// It is as yet unknown whether the remote supports the
     /// configured protocol name.
-    Unconfirmed,
+    Unknown,
     /// The configured protocol name has been confirmed by the remote
     /// but has not yet been reported to the `Kademlia` behaviour.
     Confirmed,
+    /// The configured protocol name(s) are not or no longer supported by the remote.
+    NotSupported,
     /// The configured protocol has been confirmed by the remote
     /// and the confirmation reported to the `Kademlia` behaviour.
     Reported,
@@ -226,13 +229,11 @@ impl<TUserData> InboundSubstreamState<TUserData> {
 #[derive(Debug)]
 pub enum KademliaHandlerEvent<TUserData> {
     /// The configured protocol name has been confirmed by the peer through
-    /// a successfully negotiated substream.
-    ///
-    /// This event is only emitted once by a handler upon the first
-    /// successfully negotiated inbound or outbound substream and
-    /// indicates that the connected peer participates in the Kademlia
-    /// overlay network identified by the configured protocol name.
+    /// a successfully negotiated substream or by learning the supported protocols of the remote.
     ProtocolConfirmed { endpoint: ConnectedPoint },
+    /// The configured protocol name(s) are not or no longer supported by the peer on the provided
+    /// connection and it should be removed from the routing table.
+    ProtocolNotSupported { endpoint: ConnectedPoint },
 
     /// Request for the list of nodes whose IDs are the closest to `key`. The number of nodes
     /// returned is not specified, but should be around 20.
@@ -501,7 +502,7 @@ where
             num_requested_outbound_streams: 0,
             requested_streams: Default::default(),
             keep_alive,
-            protocol_status: ProtocolStatus::Unconfirmed,
+            protocol_status: ProtocolStatus::Unknown,
         }
     }
 
@@ -520,7 +521,7 @@ where
                 protocol, msg, user_data,
             ));
         self.num_requested_outbound_streams -= 1;
-        if let ProtocolStatus::Unconfirmed = self.protocol_status {
+        if let ProtocolStatus::Unknown = 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.
@@ -542,7 +543,7 @@ where
             future::Either::Right(p) => void::unreachable(p),
         };
 
-        if let ProtocolStatus::Unconfirmed = self.protocol_status {
+        if let ProtocolStatus::Unknown = 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.
@@ -786,7 +787,28 @@ where
             ConnectionEvent::DialUpgradeError(dial_upgrade_error) => {
                 self.on_dial_upgrade_error(dial_upgrade_error)
             }
-            ConnectionEvent::AddressChange(_) | ConnectionEvent::ListenUpgradeError(_) => {}
+            ConnectionEvent::AddressChange(_)
+            | ConnectionEvent::ListenUpgradeError(_)
+            | ConnectionEvent::LocalProtocolsChange(_) => {}
+            ConnectionEvent::RemoteProtocolsChange(ProtocolsChange { protocols }) => {
+                // TODO: We should cache this / it will get simpler with #2831.
+                let kademlia_protocols = self
+                    .config
+                    .protocol_config
+                    .protocol_names()
+                    .iter()
+                    .filter_map(|b| String::from_utf8(b.to_vec()).ok())
+                    .collect::<Vec<_>>();
+
+                let remote_supports_our_kademlia_protocols =
+                    kademlia_protocols.iter().all(|p| protocols.contains(p));
+
+                if remote_supports_our_kademlia_protocols {
+                    self.protocol_status = ProtocolStatus::Confirmed;
+                } else {
+                    self.protocol_status = ProtocolStatus::NotSupported;
+                }
+            }
         }
     }
 }

protocols/kad/src/lib.rs

@@ -61,7 +61,7 @@ pub use behaviour::{
     AddProviderContext, AddProviderError, AddProviderOk, AddProviderPhase, AddProviderResult,
     BootstrapError, BootstrapOk, BootstrapResult, GetClosestPeersError, GetClosestPeersOk,
     GetClosestPeersResult, GetProvidersError, GetProvidersOk, GetProvidersResult, GetRecordError,
-    GetRecordOk, GetRecordResult, InboundRequest, NoKnownPeers, PeerRecord, PutRecordContext,
+    GetRecordOk, GetRecordResult, InboundRequest, Mode, NoKnownPeers, PeerRecord, PutRecordContext,
     PutRecordError, PutRecordOk, PutRecordPhase, PutRecordResult, QueryInfo, QueryMut, QueryRef,
     QueryResult, QueryStats,
 };