[libp2p-kad] More control & insight for k-buckets. (#1628)

* More control & insight for k-buckets.

1) More control: It is now possible to disable automatic
insertions of peers into the routing table via a new
`KademliaBucketInserts` configuration option. The
default is `OnConnected`, but it can be set to `Manual`,
in which case `add_address` must be called explicitly.
In order to communicate all situations in which a user
of `Kademlia` may want to manually update the routing
table, two new events are introduced:

  * `KademliaEvent::RoutablePeer`: When a connection to
    a peer with a known listen address is established
    which may be added to the routing table. This is
    also emitted when automatic inserts are allowed but
    the corresponding k-bucket is full.
  * `KademliaEvent::PendingRoutablePeer`: When a connection
    to a peer with a known listen address is established
    which is pending insertion into the routing table
    (but may not make it). This is only emitted when
    `OnConnected` (i.e. automatic inserts) are used.

These complement the existing `UnroutablePeer` and
`RoutingUpdated` events. It is now also possible to
explicitly remove peers and addresses from the
routing table.

2) More insight: `Kademlia::kbuckets` now gives an
iterator over `KBucketRef`s and `Kademlia::bucket`
a particular `KBucketRef`. A `KBucketRef` in turn
allows iteration over its entries. In this way,
the full contents of the routing table can be
inspected, e.g. in order to decide which peer(s)
to remove.

* Update protocols/kad/src/behaviour.rs

* Update protocols/kad/src/behaviour.rs

Co-authored-by: Max Inden <mail@max-inden.de>

* Update CHANGELOG.

Co-authored-by: Max Inden <mail@max-inden.de>

protocols/kad/src/behaviour/test.rs

@@ -196,9 +196,12 @@ fn bootstrap() {
                                 }
                                 first = false;
                                 if ok.num_remaining == 0 {
-                                    let known = swarm.kbuckets.iter()
-                                        .map(|e| e.node.key.preimage().clone())
-                                        .collect::<HashSet<_>>();
+                                    let mut known = HashSet::new();
+                                    for b in swarm.kbuckets.iter() {
+                                        for e in b.iter() {
+                                            known.insert(e.node.key.preimage().clone());
+                                        }
+                                    }
                                     assert_eq!(expected_known, known);
                                     return Poll::Ready(())
                                 }
@@ -1052,3 +1055,47 @@ fn disjoint_query_does_not_finish_before_all_paths_did() {
         record: record_trudy,
     }));
 }
+
+/// Tests that peers are not automatically inserted into
+/// the routing table with `KademliaBucketInserts::Manual`.
+#[test]
+fn manual_bucket_inserts() {
+    let mut cfg = KademliaConfig::default();
+    cfg.set_kbucket_inserts(KademliaBucketInserts::Manual);
+    // 1 -> 2 -> [3 -> ...]
+    let mut swarms = build_connected_nodes_with_config(3, 1, cfg);
+    // The peers and their addresses for which we expect `RoutablePeer` events.
+    let mut expected = swarms.iter().skip(2)
+        .map(|(a, s)| (a.clone(), Swarm::local_peer_id(s).clone()))
+        .collect::<HashMap<_,_>>();
+    // We collect the peers for which a `RoutablePeer` event
+    // was received in here to check at the end of the test
+    // that none of them was inserted into a bucket.
+    let mut routable = Vec::new();
+    // Start an iterative query from the first peer.
+    swarms[0].1.get_closest_peers(PeerId::random());
+    block_on(poll_fn(move |ctx| {
+        for (_, swarm) in swarms.iter_mut() {
+            loop {
+                match swarm.poll_next_unpin(ctx) {
+                    Poll::Ready(Some(KademliaEvent::RoutablePeer {
+                        peer, address
+                    })) => {
+                        assert_eq!(peer, expected.remove(&address).expect("Unexpected address"));
+                        routable.push(peer);
+                        if expected.is_empty() {
+                            for peer in routable.iter() {
+                                let bucket = swarm.kbucket(peer.clone()).unwrap();
+                                assert!(bucket.iter().all(|e| e.node.key.preimage() != peer));
+                            }
+                            return Poll::Ready(())
+                        }
+                    }
+                    Poll::Ready(..) => {},
+                    Poll::Pending => break
+                }
+            }
+        }
+        Poll::Pending
+    }));
+}