Somewhat complete the implementation of Kademlia records.

This commit relates to [libp2p-146] and [libp2p-1089].

  * All records expire (by default, configurable).
  * Provider records are also stored in the RecordStore, and the RecordStore
    API extended.
  * Background jobs for periodic (re-)replication and (re-)publication
    of records. Regular (value-)records are subject to re-replication and
    re-publication as per standard Kademlia. Provider records are only
    subject to re-publication.
  * For standard Kademlia value lookups (quorum = 1), the record is cached
    at the closest peer to the key that did not return the value, as per
    standard Kademlia.
  * Expiration times of regular (value-)records is computed exponentially
    inversely proportional to the number of nodes between the local node
    and the closest node known to the key (beyond the k closest), as per
    standard Kademlia.

The protobuf messages are extended with two fields: `ttl` and `publisher`
in order to implement the different semantics of re-replication (by any
of the k closest peers to the key, not affecting expiry) and re-publication
(by the original publisher, resetting the expiry). This is not done yet in
other libp2p Kademlia implementations, see e.g. [libp2p-go-323]. The new protobuf fields
have been given somewhat unique identifiers to prevent future collision.

Similarly, periodic re-publication of provider records does not seem to
be done yet in other implementations, see e.g. [libp2p-js-98].

[libp2p-146]: libp2p#146
[libp2p-1089]: libp2p#1089
[libp2p-go-323]: libp2p/go-libp2p-kad-dht#323
[libp2p-js-98]: libp2p/js-libp2p-kad-dht#98

protocols/kad/dht.proto

@@ -18,6 +18,12 @@ message Record {
 
 	// Time the record was received, set by receiver
 	string timeReceived = 5;
+
+    // The original publisher of the record.
+    bytes publisher = 6;
+
+    // The remaining TTL of the record, in seconds.
+    uint32 ttl = 7;
 };
 
 message Message {

protocols/kad/src/addresses.rs

@@ -22,67 +22,77 @@ use libp2p_core::Multiaddr;
 use smallvec::SmallVec;
 use std::fmt;
 
-/// List of addresses of a peer.
+/// A non-empty list of (unique) addresses of a peer in the routing table.
 #[derive(Clone)]
 pub struct Addresses {
     addrs: SmallVec<[Multiaddr; 6]>,
 }
 
 impl Addresses {
     /// Creates a new list of addresses.
-    pub fn new() -> Addresses {
-        Addresses {
-            addrs: SmallVec::new(),
-        }
+    pub fn new(addr: Multiaddr) -> Addresses {
+        let mut addrs = SmallVec::new();
+        addrs.push(addr);
+        Addresses { addrs }
+    }
+
+    /// Gets a reference to the first address in the list.
+    pub fn first(&self) -> &Multiaddr {
+        &self.addrs[0]
     }
 
-    /// Returns an iterator over the list of addresses.
+    /// Returns an iterator over the addresses.
     pub fn iter(&self) -> impl Iterator<Item = &Multiaddr> {
         self.addrs.iter()
     }
 
+    /// Returns the number of addresses in the list.
+    pub fn len(&self) -> usize {
+        self.addrs.len()
+    }
+
     /// Converts the addresses into a `Vec`.
     pub fn into_vec(self) -> Vec<Multiaddr> {
         self.addrs.into_vec()
     }
 
-    /// Returns true if the list of addresses is empty.
-    pub fn is_empty(&self) -> bool {
-        self.addrs.is_empty()
-    }
+    /// Removes the given address from the list.
+    ///
+    /// Returns true if the address was found and removed, false otherwise.
+    /// The last remaining address in the list cannot be remvoved.
+    ///
+    /// An address should only be removed if is determined to be invalid or
+    /// otherwise unreachable.
+    pub fn remove(&mut self, addr: &Multiaddr) -> bool {
+        if self.addrs.len() == 1 {
+            return false
+        }
 
-    /// Removes the given address from the list. Typically called if an address is determined to
-    /// be invalid or unreachable.
-    pub fn remove(&mut self, addr: &Multiaddr) {
         if let Some(pos) = self.addrs.iter().position(|a| a == addr) {
             self.addrs.remove(pos);
+            if self.addrs.len() <= self.addrs.inline_size() {
+                self.addrs.shrink_to_fit();
+            }
+            return true
         }
 
-        if self.addrs.len() <= self.addrs.inline_size() {
-            self.addrs.shrink_to_fit();
-        }
-    }
-
-    /// Clears the list. It is empty afterwards.
-    pub fn clear(&mut self) {
-        self.addrs.clear();
-        self.addrs.shrink_to_fit();
+        false
     }
 
-    /// Inserts an address in the list. No effect if the address was already in the list.
-    pub fn insert(&mut self, addr: Multiaddr) {
+    /// Adds a new address to the end of the list.
+    ///
+    /// Returns true if the address was added, false otherwise (i.e. if the
+    /// address is already in the list).
+    pub fn insert(&mut self, addr: Multiaddr) -> bool {
         if self.addrs.iter().all(|a| *a != addr) {
             self.addrs.push(addr);
+            true
+        } else {
+            false
         }
     }
 }
 
-impl Default for Addresses {
-    fn default() -> Self {
-        Addresses::new()
-    }
-}
-
 impl fmt::Debug for Addresses {
     fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
         f.debug_list()