kad: Improve robustness of addresses from the routing table (#369)

This PR improves the robustness of the kademlia addresses stored in the
routing table.

The implementation uses an eventually-consistent address store, based on
the information provided by the transport manager. This has the benefit
of not sharing the transport manager address store with the routing
table, therefore not paying for the performance hit of locking the
addresses (globally) on kademlia queries.

- transport manager exposes all dialed addresses properly to the
installed protocols
- the kademlia peer contains an address store instead of a plain vector
(bounded to 32 entries -- please note that the transport manager holds
up to 64 entries)
- similar to the transport manager, the addresses are sorted by score
depending on success / failure dials
  - when an address score is negative it is subject to removal
- on successful dial, the score is increased for the address of said
peer
- on dial failures, the score of the failed addresses is decreased
- known good addresses are always provided first
- all kademlia queries rely on this information and the addresses are
provided sorted by their score

This resolved the following issue, with the mention that undialed or
dialed but failed addresses are still provided if the address store has
enough capacity. This ensures that addresses that fail due to temporary
network errors, or errors specific to our node (ie, behind firewall --
other peers may reach the address) are provided:
- #209

cc @paritytech/networking

---------

Signed-off-by: Alexandru Vasile <[email protected]>
Co-authored-by: Dmitry Markin <[email protected]>

Author: lexnv

src/protocol/libp2p/kademlia/bucket.rs

@@ -27,7 +27,7 @@ use crate::{
 };
 
 /// K-bucket entry.
-#[derive(Debug, PartialEq, Eq)]
+#[derive(Debug)]
 pub enum KBucketEntry<'a> {
     /// Entry points to local node.
     LocalNode,
@@ -48,7 +48,7 @@ impl<'a> KBucketEntry<'a> {
         if let KBucketEntry::Vacant(old) = self {
             old.peer = new.peer;
             old.key = Key::from(new.peer);
-            old.addresses = new.addresses;
+            old.address_store = new.address_store;
             old.connection = new.connection;
         }
     }
@@ -105,7 +105,7 @@ impl KBucket {
     pub fn closest_iter<K: Clone>(&self, target: &Key<K>) -> impl Iterator<Item = &KademliaPeer> {
         let mut nodes: Vec<_> = self.nodes.iter().collect();
         nodes.sort_by(|a, b| target.distance(&a.key).cmp(&target.distance(&b.key)));
-        nodes.into_iter().filter(|peer| !peer.addresses.is_empty())
+        nodes.into_iter().filter(|peer| !peer.address_store.is_empty())
     }
 }
 

src/protocol/libp2p/kademlia/mod.rs

@@ -36,6 +36,7 @@ use crate::{
         Direction, TransportEvent, TransportService,
     },
     substream::Substream,
+    transport::Endpoint,
     types::SubstreamId,
     PeerId,
 };
@@ -220,14 +221,18 @@ impl Kademlia {
     }
 
     /// Connection established to remote peer.
-    fn on_connection_established(&mut self, peer: PeerId) -> crate::Result<()> {
+    fn on_connection_established(&mut self, peer: PeerId, endpoint: Endpoint) -> crate::Result<()> {
         tracing::trace!(target: LOG_TARGET, ?peer, "connection established");
 
         match self.peers.entry(peer) {
             Entry::Vacant(entry) => {
-                if let KBucketEntry::Occupied(entry) = self.routing_table.entry(Key::from(peer)) {
-                    entry.connection = ConnectionType::Connected;
-                }
+                // Set the conenction type to connected and potentially save the address in the
+                // table.
+                //
+                // Note: this happens regardless of the state of the kademlia managed peers, because
+                // an already occupied entry in the `self.peers` map does not mean that we are
+                // no longer interested in the address / connection type of the peer.
+                self.routing_table.on_connection_established(Key::from(peer), endpoint);
 
                 let Some(actions) = self.pending_dials.remove(&peer) else {
                     entry.insert(PeerContext::new());
@@ -262,7 +267,21 @@ impl Kademlia {
                 entry.insert(context);
                 Ok(())
             }
-            Entry::Occupied(_) => Err(Error::PeerAlreadyExists(peer)),
+            Entry::Occupied(_) => {
+                tracing::warn!(
+                    target: LOG_TARGET,
+                    ?peer,
+                    ?endpoint,
+                    "connection already exists, discarding opening substreams, this is unexpected"
+                );
+
+                // Update the connection in the routing table, similar as above. The function call
+                // happens in two places to avoid unnecessary cloning of the endpoint for logging
+                // purposes.
+                self.routing_table.on_connection_established(Key::from(peer), endpoint);
+
+                Err(Error::PeerAlreadyExists(peer))
+            }
         }
     }
 
@@ -395,12 +414,13 @@ impl Kademlia {
             .await;
 
         for info in peers {
-            self.service.add_known_address(&info.peer, info.addresses.iter().cloned());
+            let addresses = info.addresses();
+            self.service.add_known_address(&info.peer, addresses.clone().into_iter());
 
             if std::matches!(self.update_mode, RoutingTableUpdateMode::Automatic) {
                 self.routing_table.add_known_peer(
                     info.peer,
-                    info.addresses.clone(),
+                    addresses,
                     self.peers
                         .get(&info.peer)
                         .map_or(ConnectionType::NotConnected, |_| ConnectionType::Connected),
@@ -529,13 +549,15 @@ impl Kademlia {
                 );
 
                 match (providers.len(), providers.pop()) {
-                    (1, Some(provider)) =>
+                    (1, Some(provider)) => {
+                        let addresses = provider.addresses();
+
                         if provider.peer == peer {
                             self.store.put_provider(
                                 key.clone(),
                                 ContentProvider {
                                     peer,
-                                    addresses: provider.addresses.clone(),
+                                    addresses: addresses.clone(),
                                 },
                             );
 
@@ -545,7 +567,7 @@ impl Kademlia {
                                     provided_key: key,
                                     provider: ContentProvider {
                                         peer: provider.peer,
-                                        addresses: provider.addresses,
+                                        addresses,
                                     },
                                 })
                                 .await;
@@ -556,7 +578,8 @@ impl Kademlia {
                                 provider = ?provider.peer,
                                 "ignoring `ADD_PROVIDER` message with `publisher` != `provider`"
                             )
-                        },
+                        }
+                    }
                     (n, _) => {
                         tracing::trace!(
                             target: LOG_TARGET,
@@ -669,8 +692,10 @@ impl Kademlia {
     }
 
     /// Handle dial failure.
-    fn on_dial_failure(&mut self, peer: PeerId, address: Multiaddr) {
-        tracing::trace!(target: LOG_TARGET, ?peer, ?address, "failed to dial peer");
+    fn on_dial_failure(&mut self, peer: PeerId, addresses: Vec<Multiaddr>) {
+        tracing::trace!(target: LOG_TARGET, ?peer, ?addresses, "failed to dial peer");
+
+        self.routing_table.on_dial_failure(Key::from(peer), &addresses);
 
         let Some(actions) = self.pending_dials.remove(&peer) else {
             return;
@@ -682,7 +707,7 @@ impl Kademlia {
                     target: LOG_TARGET,
                     ?peer,
                     query = ?query_id,
-                    ?address,
+                    ?addresses,
                     "report failure for pending query",
                 );
 
@@ -774,7 +799,10 @@ impl Kademlia {
                     .send(KademliaEvent::FindNodeSuccess {
                         target,
                         query_id: query,
-                        peers: peers.into_iter().map(|info| (info.peer, info.addresses)).collect(),
+                        peers: peers
+                            .into_iter()
+                            .map(|info| (info.peer, info.addresses()))
+                            .collect(),
                     })
                     .await;
                 Ok(())
@@ -889,8 +917,8 @@ impl Kademlia {
 
             tokio::select! {
                 event = self.service.next() => match event {
-                    Some(TransportEvent::ConnectionEstablished { peer, .. }) => {
-                        if let Err(error) = self.on_connection_established(peer) {
+                    Some(TransportEvent::ConnectionEstablished { peer, endpoint }) => {
+                        if let Err(error) = self.on_connection_established(peer, endpoint) {
                             tracing::debug!(
                                 target: LOG_TARGET,
                                 ?error,
@@ -923,8 +951,8 @@ impl Kademlia {
                     Some(TransportEvent::SubstreamOpenFailure { substream, error }) => {
                         self.on_substream_open_failure(substream, error).await;
                     }
-                    Some(TransportEvent::DialFailure { peer, address, .. }) =>
-                        self.on_dial_failure(peer, address),
+                    Some(TransportEvent::DialFailure { peer, addresses }) =>
+                        self.on_dial_failure(peer, addresses),
                     None => return Err(Error::EssentialTaskClosed),
                 },
                 context = self.executor.next() => {
@@ -1048,7 +1076,7 @@ impl Kademlia {
 
                                 match self.routing_table.entry(Key::from(peer)) {
                                     KBucketEntry::Occupied(entry) => Some(entry.clone()),
-                                    KBucketEntry::Vacant(entry) if !entry.addresses.is_empty() =>
+                                    KBucketEntry::Vacant(entry) if !entry.address_store.is_empty() =>
                                         Some(entry.clone()),
                                     _ => None,
                                 }
@@ -1237,8 +1265,11 @@ mod tests {
             KEEP_ALIVE_TIMEOUT,
         },
         types::protocol::ProtocolName,
-        BandwidthSink,
+        BandwidthSink, ConnectionId,
     };
+    use multiaddr::Protocol;
+    use multihash::Multihash;
+    use std::str::FromStr;
     use tokio::sync::mpsc::channel;
 
     #[allow(unused)]
@@ -1379,4 +1410,89 @@ mod tests {
         // Check the local storage should not get updated.
         assert!(kademlia.store.get(&key).is_none());
     }
+
+    #[tokio::test]
+    async fn check_address_store_routing_table_updates() {
+        let (mut kademlia, _context, _manager) = make_kademlia();
+
+        let peer = PeerId::random();
+        let address_a = Multiaddr::from_str("/dns/domain1.com/tcp/30333").unwrap().with(
+            Protocol::P2p(Multihash::from_bytes(&peer.to_bytes()).unwrap()),
+        );
+        let address_b = Multiaddr::from_str("/dns/domain1.com/tcp/30334").unwrap().with(
+            Protocol::P2p(Multihash::from_bytes(&peer.to_bytes()).unwrap()),
+        );
+        let address_c = Multiaddr::from_str("/dns/domain1.com/tcp/30339").unwrap().with(
+            Protocol::P2p(Multihash::from_bytes(&peer.to_bytes()).unwrap()),
+        );
+
+        // Added only with address a.
+        kademlia.routing_table.add_known_peer(
+            peer,
+            vec![address_a.clone()],
+            ConnectionType::NotConnected,
+        );
+
+        // Check peer addresses.
+        match kademlia.routing_table.entry(Key::from(peer)) {
+            KBucketEntry::Occupied(entry) => {
+                assert_eq!(entry.addresses(), vec![address_a.clone()]);
+            }
+            _ => panic!("Peer not found in routing table"),
+        };
+
+        // Report successful connection with address b via dialer endpoint.
+        let _ = kademlia.on_connection_established(
+            peer,
+            Endpoint::Dialer {
+                address: address_b.clone(),
+                connection_id: ConnectionId::from(0),
+            },
+        );
+
+        // Address B has a higher priority, as it was detected via the dialing mechanism of the
+        // transport manager, while address A is not dialed yet.
+        match kademlia.routing_table.entry(Key::from(peer)) {
+            KBucketEntry::Occupied(entry) => {
+                assert_eq!(
+                    entry.addresses(),
+                    vec![address_b.clone(), address_a.clone()]
+                );
+            }
+            _ => panic!("Peer not found in routing table"),
+        };
+
+        // Report successful connection with a random address via listener endpoint.
+        let _ = kademlia.on_connection_established(
+            peer,
+            Endpoint::Listener {
+                address: address_c.clone(),
+                connection_id: ConnectionId::from(0),
+            },
+        );
+        // Address C was not added, as the peer has dialed us possibly on an ephemeral port.
+        match kademlia.routing_table.entry(Key::from(peer)) {
+            KBucketEntry::Occupied(entry) => {
+                assert_eq!(
+                    entry.addresses(),
+                    vec![address_b.clone(), address_a.clone()]
+                );
+            }
+            _ => panic!("Peer not found in routing table"),
+        };
+
+        // Address B fails two times (which gives it a lower score than A) and
+        // makes it subject to removal.
+        kademlia.on_dial_failure(peer, vec![address_b.clone(), address_b.clone()]);
+
+        match kademlia.routing_table.entry(Key::from(peer)) {
+            KBucketEntry::Occupied(entry) => {
+                assert_eq!(
+                    entry.addresses(),
+                    vec![address_a.clone(), address_b.clone()]
+                );
+            }
+            _ => panic!("Peer not found in routing table"),
+        };
+    }
 }

src/protocol/libp2p/kademlia/query/find_node.rs

@@ -317,7 +317,7 @@ mod tests {
         KademliaPeer {
             peer,
             key: Key::from(peer),
-            addresses: vec![],
+            address_store: Default::default(),
             connection: ConnectionType::Connected,
         }
     }
@@ -353,7 +353,12 @@ mod tests {
         let mut context = FindNodeContext::new(config, VecDeque::new());
         assert!(context.is_done());
         let event = context.next_action().unwrap();
-        assert_eq!(event, QueryAction::QueryFailed { query: QueryId(0) });
+        match event {
+            QueryAction::QueryFailed { query, .. } => {
+                assert_eq!(query, QueryId(0));
+            }
+            _ => panic!("Unexpected event"),
+        };
     }
 
     #[test]
@@ -522,7 +527,12 @@ mod tests {
 
         // Produces the result.
         let event = context.next_action().unwrap();
-        assert_eq!(event, QueryAction::QuerySucceeded { query: QueryId(0) });
+        match event {
+            QueryAction::QuerySucceeded { query, .. } => {
+                assert_eq!(query, QueryId(0));
+            }
+            _ => panic!("Unexpected event"),
+        };
     }
 
     #[test]
@@ -550,7 +560,12 @@ mod tests {
         context.register_response(closest, vec![]);
 
         let event = context.next_action().unwrap();
-        assert_eq!(event, QueryAction::QuerySucceeded { query: QueryId(0) });
+        match event {
+            QueryAction::QuerySucceeded { query } => {
+                assert_eq!(query, QueryId(0));
+            }
+            _ => panic!("Unexpected event"),
+        };
     }
 
     #[test]
@@ -602,7 +617,12 @@ mod tests {
         context.register_response(closest, vec![]);
 
         let event = context.next_action().unwrap();
-        assert_eq!(event, QueryAction::QuerySucceeded { query: QueryId(0) });
+        match event {
+            QueryAction::QuerySucceeded { query, .. } => {
+                assert_eq!(query, QueryId(0));
+            }
+            _ => panic!("Unexpected event"),
+        };
     }
 
     #[test]
@@ -665,7 +685,12 @@ mod tests {
 
         // Produces the result.
         let event = context.next_action().unwrap();
-        assert_eq!(event, QueryAction::QuerySucceeded { query: QueryId(0) });
+        match event {
+            QueryAction::QuerySucceeded { query } => {
+                assert_eq!(query, QueryId(0));
+            }
+            _ => panic!("Unexpected event"),
+        };
 
         // Because the FindNode query keeps a window of the best K (3 in this case) peers,
         // we expect to produce the best K peers. As opposed to having only the last entry