script/research/dhtd: replaced String with blake3::Hash

script/research/dhtd/src/dht.rs

@@ -29,7 +29,6 @@ pub type DhtPtr = Arc<RwLock<Dht>>;
 
 // TODO: proper errors
 // TODO: lookup table to be based on directly connected peers, not broadcast based
-// TODO: replace Strings with blake3 hashes
 // Using string in structures because we are at an external crate
 // and cant use blake3 serialization. To be replaced once merged with core src.
 
@@ -38,9 +37,9 @@ pub struct Dht {
     /// Daemon id
     pub id: blake3::Hash,
     /// Daemon hasmap
-    pub map: FxHashMap<String, Vec<u8>>,
+    pub map: FxHashMap<blake3::Hash, Vec<u8>>,
     /// Network lookup map, containing nodes that holds each key
-    pub lookup: FxHashMap<String, HashSet<String>>,
+    pub lookup: FxHashMap<blake3::Hash, HashSet<blake3::Hash>>,
     /// P2P network pointer
     p2p: P2pPtr,
     /// Channel to receive responses from P2P
@@ -49,12 +48,12 @@ pub struct Dht {
     stop_signal: async_channel::Receiver<()>,
     /// Daemon seen requests/responses ids and timestamp,
     /// to prevent rebroadcasting and loops
-    pub seen: FxHashMap<String, i64>,
+    pub seen: FxHashMap<blake3::Hash, i64>,
 }
 
 impl Dht {
     pub async fn new(
-        initial: Option<FxHashMap<String, HashSet<String>>>,
+        initial: Option<FxHashMap<blake3::Hash, HashSet<blake3::Hash>>>,
         p2p_ptr: P2pPtr,
         stop_signal: async_channel::Receiver<()>,
         ex: Arc<Executor<'_>>,
@@ -99,33 +98,52 @@ impl Dht {
         Ok(dht)
     }
 
-    /// Store provided key value pair and update lookup map
-    pub async fn insert(&mut self, key: String, value: Vec<u8>) -> Result<Option<String>> {
+    /// Store provided key value pair, update lookup map and broadcast new insert to network
+    pub async fn insert(
+        &mut self,
+        key: blake3::Hash,
+        value: Vec<u8>,
+    ) -> Result<Option<blake3::Hash>> {
         self.map.insert(key.clone(), value);
-        self.lookup_insert(key, self.id.to_string()).await
+
+        if let Err(e) = self.lookup_insert(key, self.id) {
+            error!("Failed to insert record to lookup map: {}", e);
+            return Err(e)
+        };
+
+        let request = LookupRequest::new(self.id, key.clone(), 0);
+        if let Err(e) = self.p2p.broadcast(request).await {
+            error!("Failed broadcasting request: {}", e);
+            return Err(e)
+        }
+
+        Ok(Some(key))
     }
 
     /// Remove provided key value pair and update lookup map
-    pub async fn remove(&mut self, key: String) -> Result<Option<String>> {
+    pub async fn remove(&mut self, key: blake3::Hash) -> Result<Option<blake3::Hash>> {
         // Check if key value pair existed and act accordingly
         match self.map.remove(&key) {
             Some(_) => {
                 debug!("Key removed: {}", key);
-                let daemon = self.id.to_string();
-                let request = LookupRequest::new(daemon, key.clone(), 1);
+                let request = LookupRequest::new(self.id, key.clone(), 1);
                 if let Err(e) = self.p2p.broadcast(request).await {
                     error!("Failed broadcasting request: {}", e);
                     return Err(e)
                 }
 
-                self.lookup_remove(key.clone(), self.id.to_string())
+                self.lookup_remove(key.clone(), self.id)
             }
             None => Ok(None),
         }
     }
 
     /// Store provided key node pair in lookup map and update network
-    pub async fn lookup_insert(&mut self, key: String, node_id: String) -> Result<Option<String>> {
+    pub fn lookup_insert(
+        &mut self,
+        key: blake3::Hash,
+        node_id: blake3::Hash,
+    ) -> Result<Option<blake3::Hash>> {
         let mut lookup_set = match self.lookup.get(&key) {
             Some(s) => s.clone(),
             None => HashSet::new(),
@@ -134,18 +152,15 @@ impl Dht {
         lookup_set.insert(node_id);
         self.lookup.insert(key.clone(), lookup_set);
 
-        let daemon = self.id.to_string();
-        let request = LookupRequest::new(daemon, key.clone(), 0);
-        if let Err(e) = self.p2p.broadcast(request).await {
-            error!("Failed broadcasting request: {}", e);
-            return Err(e)
-        }
-
         Ok(Some(key))
     }
 
     /// Remove provided node id from keys set in local lookup map
-    pub fn lookup_remove(&mut self, key: String, node_id: String) -> Result<Option<String>> {
+    pub fn lookup_remove(
+        &mut self,
+        key: blake3::Hash,
+        node_id: blake3::Hash,
+    ) -> Result<Option<blake3::Hash>> {
         if let Some(s) = self.lookup.get(&key) {
             let mut lookup_set = s.clone();
             lookup_set.remove(&node_id);
@@ -160,7 +175,7 @@ impl Dht {
     }
 
     /// Verify if provided key exists and return flag if local or in network
-    pub fn contains_key(&self, key: String) -> Option<bool> {
+    pub fn contains_key(&self, key: blake3::Hash) -> Option<bool> {
         match self.lookup.contains_key(&key) {
             true => Some(self.map.contains_key(&key)),
             false => None,
@@ -168,12 +183,12 @@ impl Dht {
     }
 
     /// Get key from local map, acting as daemon cache
-    pub fn get(&self, key: String) -> Option<&Vec<u8>> {
+    pub fn get(&self, key: blake3::Hash) -> Option<&Vec<u8>> {
         self.map.get(&key)
     }
 
     /// Generate key request and broadcast it to the network
-    pub async fn request_key(&self, key: String) -> Result<()> {
+    pub async fn request_key(&self, key: blake3::Hash) -> Result<()> {
         // Verify the key exist in the lookup map.
         let peers = match self.lookup.get(&key) {
             Some(v) => v.clone(),
@@ -195,10 +210,9 @@ impl Dht {
         }
 
         // We create a key request, and broadcast it to the network
-        let daemon = self.id.to_string();
         // We choose last known peer as request recipient
-        let peer = peers.iter().last().unwrap().to_string();
-        let request = KeyRequest::new(daemon.clone(), peer, key.clone());
+        let peer = peers.iter().last().unwrap().clone();
+        let request = KeyRequest::new(self.id, peer, key);
         // TODO: ask connected peers directly, not broadcast
         if let Err(e) = self.p2p.broadcast(request).await {
             error!("Failed broadcasting request: {}", e);

script/research/dhtd/src/main.rs

@@ -21,6 +21,7 @@ use darkfi::{
     util::{
         cli::{get_log_config, get_log_level, spawn_config},
         path::get_config_path,
+        serial::serialize,
     },
     Result,
 };
@@ -99,10 +100,11 @@ impl Dhtd {
         }
 
         let key = params[0].to_string();
+        let key_hash = blake3::hash(&serialize(&key));
 
         // We execute this sequence to prevent lock races between threads
         // Verify key exists
-        let exists = self.dht.read().await.contains_key(key.clone());
+        let exists = self.dht.read().await.contains_key(key_hash.clone());
         if let None = exists {
             info!("Did not find key: {}", key);
             return server_error(RpcError::UnknownKey, id).into()
@@ -111,7 +113,7 @@ impl Dhtd {
         // Check if key is local or shoud query network
         let local = exists.unwrap();
         if local {
-            match self.dht.read().await.get(key.clone()) {
+            match self.dht.read().await.get(key_hash.clone()) {
                 Some(value) => {
                     let string = std::str::from_utf8(&value).unwrap().to_string();
                     return JsonResponse::new(json!((key, string)), id).into()
@@ -124,7 +126,7 @@ impl Dhtd {
         }
 
         info!("Key doesn't exist locally, querring network...");
-        if let Err(e) = self.dht.read().await.request_key(key.clone()).await {
+        if let Err(e) = self.dht.read().await.request_key(key_hash).await {
             error!("Failed to query key: {}", e);
             return server_error(RpcError::QueryFailed, id).into()
         }
@@ -168,10 +170,10 @@ impl Dhtd {
         }
 
         let key = params[0].to_string();
+        let key_hash = blake3::hash(&serialize(&key));
         let value = params[1].to_string();
 
-        if let Err(e) = self.dht.write().await.insert(key.clone(), value.as_bytes().to_vec()).await
-        {
+        if let Err(e) = self.dht.write().await.insert(key_hash, value.as_bytes().to_vec()).await {
             error!("Failed to insert key: {}", e);
             return server_error(RpcError::KeyInsertFail, id)
         }
@@ -189,13 +191,15 @@ impl Dhtd {
         }
 
         let key = params[0].to_string();
+        let key_hash = blake3::hash(&serialize(&key));
+
         // Check if key value pair existed and act accordingly
-        let result = self.dht.write().await.remove(key.clone()).await;
+        let result = self.dht.write().await.remove(key_hash).await;
         match result {
             Ok(option) => match option {
                 Some(k) => {
-                    info!("Key removed: {}", k);
-                    JsonResponse::new(json!(k), id).into()
+                    info!("Hash key removed: {}", k);
+                    JsonResponse::new(json!(k.to_string()), id).into()
                 }
                 None => {
                     info!("Did not find key: {}", key);
@@ -215,7 +219,8 @@ impl Dhtd {
     // <-- {"jsonrpc": "2.0", "result": "map", "id": 1}
     pub async fn map(&self, id: Value, _params: &[Value]) -> JsonResult {
         let map = self.dht.read().await.map.clone();
-        JsonResponse::new(json!(map), id).into()
+        let map_string = format!("{:#?}", map);
+        JsonResponse::new(json!(map_string), id).into()
     }
 
     // RPCAPI:
@@ -224,7 +229,8 @@ impl Dhtd {
     // <-- {"jsonrpc": "2.0", "result": "lookup", "id": 1}
     pub async fn lookup(&self, id: Value, _params: &[Value]) -> JsonResult {
         let lookup = self.dht.read().await.lookup.clone();
-        JsonResponse::new(json!(lookup), id).into()
+        let lookup_string = format!("{:#?}", lookup);
+        JsonResponse::new(json!(lookup_string), id).into()
     }
 }
 

script/research/dhtd/src/messages.rs

@@ -9,21 +9,21 @@ use darkfi::{
 #[derive(Debug, Clone, SerialDecodable, SerialEncodable)]
 pub struct KeyRequest {
     /// Request id    
-    pub id: String,
+    pub id: blake3::Hash,
     /// Daemon id requesting the key
-    pub from: String,
+    pub from: blake3::Hash,
     /// Daemon id holding the key
-    pub to: String,
+    pub to: blake3::Hash,
     /// Key entry
-    pub key: String,
+    pub key: blake3::Hash,
 }
 
 impl KeyRequest {
-    pub fn new(from: String, to: String, key: String) -> Self {
+    pub fn new(from: blake3::Hash, to: blake3::Hash, key: blake3::Hash) -> Self {
         // Generate a random id
         let mut rng = rand::thread_rng();
         let n: u16 = rng.gen();
-        let id = blake3::hash(&serialize(&n)).to_string();
+        let id = blake3::hash(&serialize(&n));
         Self { id, from, to, key }
     }
 }
@@ -38,23 +38,23 @@ impl net::Message for KeyRequest {
 #[derive(Debug, Clone, SerialDecodable, SerialEncodable)]
 pub struct KeyResponse {
     /// Response id
-    pub id: String,
+    pub id: blake3::Hash,
     /// Daemon id holding the key
-    pub from: String,
+    pub from: blake3::Hash,
     /// Daemon id holding the key
-    pub to: String,
+    pub to: blake3::Hash,
     /// Key entry
-    pub key: String,
+    pub key: blake3::Hash,
     /// Key value
     pub value: Vec<u8>,
 }
 
 impl KeyResponse {
-    pub fn new(from: String, to: String, key: String, value: Vec<u8>) -> Self {
+    pub fn new(from: blake3::Hash, to: blake3::Hash, key: blake3::Hash, value: Vec<u8>) -> Self {
         // Generate a random id
         let mut rng = rand::thread_rng();
         let n: u16 = rng.gen();
-        let id = blake3::hash(&serialize(&n)).to_string();
+        let id = blake3::hash(&serialize(&n));
         Self { id, from, to, key, value }
     }
 }
@@ -69,21 +69,21 @@ impl net::Message for KeyResponse {
 #[derive(Debug, Clone, SerialDecodable, SerialEncodable)]
 pub struct LookupRequest {
     /// Request id    
-    pub id: String,
+    pub id: blake3::Hash,
     /// Daemon id executing the request
-    pub daemon: String,
+    pub daemon: blake3::Hash,
     /// Key entry
-    pub key: String,
+    pub key: blake3::Hash,
     /// Request type
     pub req_type: u8, // 0 for insert, 1 for remove
 }
 
 impl LookupRequest {
-    pub fn new(daemon: String, key: String, req_type: u8) -> Self {
+    pub fn new(daemon: blake3::Hash, key: blake3::Hash, req_type: u8) -> Self {
         // Generate a random id
         let mut rng = rand::thread_rng();
         let n: u16 = rng.gen();
-        let id = blake3::hash(&serialize(&n)).to_string();
+        let id = blake3::hash(&serialize(&n));
         Self { id, daemon, key, req_type }
     }
 }

script/research/dhtd/src/protocol.rs

@@ -84,7 +84,7 @@ impl Protocol {
                 dht.seen.insert(req_copy.id.clone(), Utc::now().timestamp());
             }
 
-            let daemon = self.dht.read().await.id.to_string();
+            let daemon = self.dht.read().await.id;
             if daemon != req_copy.to {
                 if let Err(e) =
                     self.p2p.broadcast_with_exclude(req_copy.clone(), &exclude_list).await
@@ -137,7 +137,7 @@ impl Protocol {
                 dht.seen.insert(resp_copy.id.clone(), Utc::now().timestamp());
             }
 
-            if self.dht.read().await.id.to_string() != resp_copy.to {
+            if self.dht.read().await.id != resp_copy.to {
                 if let Err(e) =
                     self.p2p.broadcast_with_exclude(resp_copy.clone(), &exclude_list).await
                 {
@@ -183,13 +183,11 @@ impl Protocol {
             }
 
             let result = match req_copy.req_type {
-                0 => {
-                    self.dht
-                        .write()
-                        .await
-                        .lookup_insert(req_copy.key.clone(), req_copy.daemon.clone())
-                        .await
-                }
+                0 => self
+                    .dht
+                    .write()
+                    .await
+                    .lookup_insert(req_copy.key.clone(), req_copy.daemon.clone()),
                 _ => self
                     .dht
                     .write()