kad: Refactor GetRecord query and add tests

src/protocol/libp2p/kademlia/handle.rs

@@ -153,8 +153,8 @@ pub enum KademliaEvent {
         /// Query ID.
         query_id: QueryId,
 
-        /// Found record.
-        record: PeerRecord,
+        /// Found records.
+        records: RecordsType,
     },
 
     /// `PUT_VALUE` query succeeded.
@@ -173,6 +173,18 @@ pub enum KademliaEvent {
     },
 }
 
+/// The type of the DHT records.
+#[derive(Debug, Clone)]
+pub enum RecordsType {
+    /// Record was found in the local store.
+    ///
+    /// This contains only a single result.
+    LocalStore(Record),
+
+    /// Records found in the network.
+    Network(Vec<PeerRecord>),
+}
+
 /// Handle for communicating with the Kademlia protocol.
 pub struct KademliaHandle {
     /// TX channel for sending commands to `Kademlia`.

src/protocol/libp2p/kademlia/mod.rs

@@ -52,6 +52,8 @@ pub use handle::{KademliaEvent, KademliaHandle, Quorum, RoutingTableUpdateMode};
 pub use query::QueryId;
 pub use record::{Key as RecordKey, PeerRecord, Record};
 
+use self::handle::RecordsType;
+
 /// Logging target for the file.
 const LOG_TARGET: &str = "litep2p::ipfs::kademlia";
 
@@ -636,11 +638,32 @@ impl Kademlia {
 
                 Ok(())
             }
-            QueryAction::GetRecordQueryDone { query_id, record } => {
-                self.store.put(record.record.clone());
+            QueryAction::GetRecordQueryDone { query_id, records } => {
+                // Considering this gives a view of all peers and their records, some peers may have
+                // outdated records. Store only the record which is backed by most
+                // peers.
+                let rec = records
+                    .iter()
+                    .map(|peer_record| &peer_record.record)
+                    .fold(HashMap::new(), |mut acc, rec| {
+                        *acc.entry(rec).or_insert(0) += 1;
+                        acc
+                    })
+                    .into_iter()
+                    .max_by_key(|(_, v)| *v)
+                    .map(|(k, _)| k);
+
+                if let Some(record) = rec {
+                    self.store.put(record.clone());
+                }
 
-                let _ =
-                    self.event_tx.send(KademliaEvent::GetRecordSuccess { query_id, record }).await;
+                let _ = self
+                    .event_tx
+                    .send(KademliaEvent::GetRecordSuccess {
+                        query_id,
+                        records: RecordsType::Network(records),
+                    })
+                    .await;
                 Ok(())
             }
             QueryAction::QueryFailed { query } => {
@@ -782,7 +805,7 @@ impl Kademlia {
                                 (Some(record), Quorum::One) => {
                                     let _ = self
                                         .event_tx
-                                        .send(KademliaEvent::GetRecordSuccess { query_id, record: PeerRecord { record: record.clone(), peer: None } })
+                                        .send(KademliaEvent::GetRecordSuccess { query_id, records: RecordsType::LocalStore(record.clone()) })
                                         .await;
                                 }
                                 (record, _) => {