protocols/kad/behaviour: Remove false assert on connected_peers (#2120)

Given the following scenario:

1. Remote peer X connects and is added to `connected_peers`.
2. Remote peer X opens a Kademlia substream and thus confirms that it supports
   the Kademlia protocol.
3. Remote peer X is added to the routing table as `Connected`.
4. Remote peer X disconnects and is thus marked as `Disconnected` in the routing
   table.
5. Remote peer Y connects and is added to `connected_peers`.
6. Remote peer X re-connects and is added to `connected_peers`.
7. Remote peer Y opens a Kademlia substream and thus confirms that it supports
   the Kademlia protocol.
8. Remote peer Y is added to the routing table. Given that the bucket is already
   full the call to `entry.insert` returns `kbucket::InsertResult::Pending {
   disconnected }` where disconnected is peer X.

While peer X is in `connected_peers` it has not yet (re-) confirmed that it
supports the Kademlia routing protocol and thus is still tracked as
`Disconnected` in the routing table. The `debug_assert` removed in this pull
request does not capture this scenario.

protocols/kad/CHANGELOG.md

@@ -11,7 +11,10 @@
 
 - Expose kbucket range on `KademliaEvent::RoutingUpdated` (see [PR 2087]).
 
+- Remove false `debug_assert` on `connected_peers` (see [PR 2120]).
+
 [PR 2087]: https://github.com/libp2p/rust-libp2p/pull/2087
+[PR 2120]: https://github.com/libp2p/rust-libp2p/pull/2120
 
 # 0.30.0 [2021-04-13]
 

protocols/kad/src/behaviour.rs

@@ -1042,15 +1042,23 @@ where
                                 ));
                             },
                             kbucket::InsertResult::Pending { disconnected } => {
-                                debug_assert!(!self.connected_peers.contains(disconnected.preimage()));
                                 let address = addresses.first().clone();
                                 self.queued_events.push_back(NetworkBehaviourAction::GenerateEvent(
                                     KademliaEvent::PendingRoutablePeer { peer, address }
                                 ));
-                                self.queued_events.push_back(NetworkBehaviourAction::DialPeer {
-                                    peer_id: disconnected.into_preimage(),
-                                    condition: DialPeerCondition::Disconnected
-                                })
+
+                                // `disconnected` might already be in the process of re-connecting.
+                                // In other words `disconnected` might have already re-connected but
+                                // is not yet confirmed to support the Kademlia protocol via
+                                // [`KademliaHandlerEvent::ProtocolConfirmed`].
+                                //
+                                // Only try dialing peer if not currently connected.
+                                if !self.connected_peers.contains(disconnected.preimage()) {
+                                    self.queued_events.push_back(NetworkBehaviourAction::DialPeer {
+                                        peer_id: disconnected.into_preimage(),
+                                        condition: DialPeerCondition::Disconnected
+                                    })
+                                }
                             },
                         }
                     }