Implement portal_history* add_enr, get_enr, delete_enr, lookup_enr RPC's (#718)

* Implementing the last 4 unimplemented portal_history* RPC's

* Added peertests for history* ``addEnr``, ``getEnr``, ``deleteEnr``, ``lookupEnr`` RPC's

Author: KolbyML

ethportal-api/src/discv5.rs

@@ -34,7 +34,7 @@ pub trait Discv5Api {
     #[method(name = "deleteEnr")]
     async fn delete_enr(&self, node_id: NodeId) -> RpcResult<bool>;
 
-    /// Fetch the ENR representation associated with the given Node ID and optional sequence number.
+    /// Fetch the ENR representation associated with the given Node ID.
     #[method(name = "lookupEnr")]
-    async fn lookup_enr(&self, node_id: NodeId, enr_seq: Option<u32>) -> RpcResult<Enr>;
+    async fn lookup_enr(&self, node_id: NodeId) -> RpcResult<Enr>;
 }

ethportal-api/src/history.rs

@@ -31,9 +31,9 @@ pub trait HistoryNetworkApi {
     #[method(name = "historyDeleteEnr")]
     async fn delete_enr(&self, node_id: NodeId) -> RpcResult<bool>;
 
-    /// Fetch the ENR representation associated with the given Node ID and optional sequence number.
+    /// Fetch the ENR representation associated with the given Node ID.
     #[method(name = "historyLookupEnr")]
-    async fn lookup_enr(&self, node_id: NodeId, enr_seq: Option<u32>) -> RpcResult<Enr>;
+    async fn lookup_enr(&self, node_id: NodeId) -> RpcResult<Enr>;
 
     /// Send a PING message to the designated node and wait for a PONG response
     #[method(name = "historyPing")]

ethportal-peertest/src/scenarios/basic.rs

@@ -42,6 +42,42 @@ pub async fn test_history_radius(target: &Client) {
     );
 }
 
+pub async fn test_history_add_enr(target: &Client, peertest: &Peertest) {
+    info!("Testing portal_historyAddEnr");
+    let result = HistoryNetworkApiClient::add_enr(target, peertest.bootnode.enr.clone())
+        .await
+        .unwrap();
+    assert!(result);
+}
+
+pub async fn test_history_get_enr(target: &Client, peertest: &Peertest) {
+    info!("Testing portal_historyGetEnr");
+    let result = HistoryNetworkApiClient::get_enr(target, peertest.bootnode.enr.node_id().into())
+        .await
+        .unwrap();
+    assert_eq!(result, peertest.bootnode.enr);
+}
+
+pub async fn test_history_delete_enr(target: &Client, peertest: &Peertest) {
+    info!("Testing portal_historyDeleteEnr");
+    let result =
+        HistoryNetworkApiClient::delete_enr(target, peertest.bootnode.enr.node_id().into())
+            .await
+            .unwrap();
+    assert!(result);
+}
+
+pub async fn test_history_lookup_enr(peertest: &Peertest) {
+    info!("Testing portal_historyLookupEnr");
+    let result = HistoryNetworkApiClient::lookup_enr(
+        &peertest.bootnode.ipc_client,
+        peertest.nodes[0].enr.node_id().into(),
+    )
+    .await
+    .unwrap();
+    assert_eq!(result, peertest.nodes[0].enr);
+}
+
 pub async fn test_history_ping(target: &Client, peertest: &Peertest) {
     info!("Testing portal_historyPing");
     let result = target.ping(peertest.bootnode.enr.clone()).await.unwrap();

newsfragments/718.added.md

@@ -0,0 +1 @@
+Implement portal_history* ``add_enr``, ``get_enr``, ``delete_enr``, ``lookup_enr`` RPC's

portalnet/src/overlay.rs

@@ -1,19 +1,21 @@
+use discv5::{
+    enr::NodeId,
+    kbucket::{
+        Entry, FailureReason, Filter, InsertResult, KBucketsTable, Key, NodeStatus,
+        MAX_NODES_PER_BUCKET,
+    },
+    ConnectionDirection, ConnectionState, TalkRequest,
+};
+use futures::channel::oneshot;
+use parking_lot::RwLock;
+use ssz::Encode;
 use std::{
     collections::{BTreeMap, HashSet},
     fmt::{Debug, Display},
     marker::{PhantomData, Sync},
     sync::Arc,
     time::Duration,
 };
-
-use discv5::{
-    enr::NodeId,
-    kbucket::{Filter, KBucketsTable, NodeStatus, MAX_NODES_PER_BUCKET},
-    TalkRequest,
-};
-use futures::channel::oneshot;
-use parking_lot::RwLock;
-use ssz::Encode;
 use tokio::sync::mpsc::UnboundedSender;
 use tracing::{debug, error, info, warn};
 use utp_rs::socket::UtpSocket;
@@ -276,6 +278,97 @@ where
             .collect()
     }
 
+    /// `AddEnr` adds requested `enr` to our kbucket.
+    pub fn add_enr(&self, enr: Enr) -> Result<(), OverlayRequestError> {
+        let key = Key::from(enr.node_id());
+        match self.kbuckets.write().insert_or_update(
+            &key,
+            Node {
+                enr,
+                data_radius: Distance::MAX,
+            },
+            NodeStatus {
+                state: ConnectionState::Disconnected,
+                direction: ConnectionDirection::Incoming,
+            },
+        ) {
+            InsertResult::Inserted
+            | InsertResult::Pending { .. }
+            | InsertResult::StatusUpdated { .. }
+            | InsertResult::ValueUpdated
+            | InsertResult::Updated { .. }
+            | InsertResult::UpdatedPending => Ok(()),
+            InsertResult::Failed(FailureReason::BucketFull) => {
+                Err(OverlayRequestError::Failure("The bucket was full.".into()))
+            }
+            InsertResult::Failed(FailureReason::BucketFilter) => Err(OverlayRequestError::Failure(
+                "The node didn't pass the bucket filter.".into(),
+            )),
+            InsertResult::Failed(FailureReason::TableFilter) => Err(OverlayRequestError::Failure(
+                "The node didn't pass the table filter.".into(),
+            )),
+            InsertResult::Failed(FailureReason::InvalidSelfUpdate) => {
+                Err(OverlayRequestError::Failure("Cannot update self.".into()))
+            }
+            InsertResult::Failed(_) => {
+                Err(OverlayRequestError::Failure("Failed to insert ENR".into()))
+            }
+        }
+    }
+
+    /// `GetEnr` gets requested `enr` from our kbucket.
+    pub fn get_enr(&self, node_id: NodeId) -> Result<Enr, OverlayRequestError> {
+        if node_id == self.local_enr().node_id() {
+            return Ok(self.local_enr());
+        }
+        let key = Key::from(node_id);
+        if let Entry::Present(entry, _) = self.kbuckets.write().entry(&key) {
+            return Ok(entry.value().enr());
+        }
+        Err(OverlayRequestError::Failure("Couldn't get ENR".into()))
+    }
+
+    /// `DeleteEnr` deletes requested `enr` from our kbucket.
+    pub fn delete_enr(&self, node_id: NodeId) -> bool {
+        let key = &Key::from(node_id);
+        self.kbuckets.write().remove(key)
+    }
+
+    /// `LookupEnr` finds requested `enr` from our kbucket, FindNode, and RecursiveFindNode.
+    pub async fn lookup_enr(&self, node_id: NodeId) -> Result<Enr, OverlayRequestError> {
+        if node_id == self.local_enr().node_id() {
+            return Ok(self.local_enr());
+        }
+
+        let enr = self.get_enr(node_id);
+
+        // try to find more up to date enr
+        if let Ok(enr) = enr.clone() {
+            let nodes = self.send_find_nodes(enr, vec![0]).await?;
+            let enr_highest_seq = nodes.enrs.into_iter().max_by(|a, b| a.seq().cmp(&b.seq()));
+
+            if let Some(enr_highest_seq) = enr_highest_seq {
+                return Ok(enr_highest_seq.into());
+            }
+        }
+
+        let lookup_node_enr = self.lookup_node(node_id).await;
+        let lookup_node_enr = lookup_node_enr
+            .into_iter()
+            .max_by(|a, b| a.seq().cmp(&b.seq()));
+        if let Some(lookup_node_enr) = lookup_node_enr {
+            let mut enr_seq = 0;
+            if let Ok(enr) = enr.clone() {
+                enr_seq = enr.seq();
+            }
+            if lookup_node_enr.seq() > enr_seq {
+                return Ok(lookup_node_enr);
+            }
+        }
+
+        enr
+    }
+
     /// Sends a `Ping` request to `enr`.
     pub async fn send_ping(&self, enr: Enr) -> Result<Pong, OverlayRequestError> {
         // Construct the request.

rpc/src/discv5_rpc.rs

@@ -52,8 +52,8 @@ impl Discv5ApiServer for Discv5Api {
         Err(Error::MethodNotFound("delete_enr".to_owned()))
     }
 
-    /// Fetch the ENR representation associated with the given Node ID and optional sequence number.
-    async fn lookup_enr(&self, _node_id: NodeId, _enr_seq: Option<u32>) -> RpcResult<Enr> {
+    /// Fetch the ENR representation associated with the given Node ID.
+    async fn lookup_enr(&self, _node_id: NodeId) -> RpcResult<Enr> {
         Err(Error::MethodNotFound("lookup_enr".to_owned()))
     }
 }

rpc/src/history_rpc.rs

@@ -1,4 +1,4 @@
-use crate::jsonrpsee::core::{async_trait, Error, RpcResult};
+use crate::jsonrpsee::core::{async_trait, RpcResult};
 use anyhow::anyhow;
 use ethportal_api::types::portal::{
     AcceptInfo, ContentInfo, DataRadius, FindNodesInfo, PaginateLocalContentInfo, PongInfo,
@@ -59,23 +59,35 @@ impl HistoryNetworkApiServer for HistoryNetworkApi {
     }
 
     /// Write an Ethereum Node Record to the overlay routing table.
-    async fn add_enr(&self, _enr: Enr) -> RpcResult<bool> {
-        Err(Error::MethodNotFound("add_enr".to_owned()))
+    async fn add_enr(&self, enr: Enr) -> RpcResult<bool> {
+        let endpoint = HistoryEndpoint::AddEnr(enr);
+        let result = self.proxy_query_to_history_subnet(endpoint).await?;
+        let result: bool = from_value(result)?;
+        Ok(result)
     }
 
     /// Fetch the latest ENR associated with the given node ID.
-    async fn get_enr(&self, _node_id: NodeId) -> RpcResult<Enr> {
-        Err(Error::MethodNotFound("get_enr".to_owned()))
+    async fn get_enr(&self, node_id: NodeId) -> RpcResult<Enr> {
+        let endpoint = HistoryEndpoint::GetEnr(node_id);
+        let result = self.proxy_query_to_history_subnet(endpoint).await?;
+        let result: Enr = from_value(result)?;
+        Ok(result)
     }
 
     /// Delete Node ID from the overlay routing table.
-    async fn delete_enr(&self, _node_id: NodeId) -> RpcResult<bool> {
-        Err(Error::MethodNotFound("delete_enr".to_owned()))
+    async fn delete_enr(&self, node_id: NodeId) -> RpcResult<bool> {
+        let endpoint = HistoryEndpoint::DeleteEnr(node_id);
+        let result = self.proxy_query_to_history_subnet(endpoint).await?;
+        let result: bool = from_value(result)?;
+        Ok(result)
     }
 
-    /// Fetch the ENR representation associated with the given Node ID and optional sequence number.
-    async fn lookup_enr(&self, _node_id: NodeId, _enr_seq: Option<u32>) -> RpcResult<Enr> {
-        Err(Error::MethodNotFound("lookup_enr".to_owned()))
+    /// Fetch the ENR representation associated with the given Node ID.
+    async fn lookup_enr(&self, node_id: NodeId) -> RpcResult<Enr> {
+        let endpoint = HistoryEndpoint::LookupEnr(node_id);
+        let result = self.proxy_query_to_history_subnet(endpoint).await?;
+        let result: Enr = from_value(result)?;
+        Ok(result)
     }
 
     /// Send a PING message to the designated node and wait for a PONG response

tests/self_peertest.rs

@@ -59,6 +59,10 @@ mod test {
         peertest::scenarios::basic::test_discv5_node_info(&peertest).await;
         peertest::scenarios::basic::test_discv5_routing_table_info(&target).await;
         peertest::scenarios::basic::test_history_radius(&target).await;
+        peertest::scenarios::basic::test_history_add_enr(&target, &peertest).await;
+        peertest::scenarios::basic::test_history_get_enr(&target, &peertest).await;
+        peertest::scenarios::basic::test_history_delete_enr(&target, &peertest).await;
+        peertest::scenarios::basic::test_history_lookup_enr(&peertest).await;
         peertest::scenarios::basic::test_history_ping(&target, &peertest).await;
         peertest::scenarios::basic::test_history_find_nodes(&target, &peertest).await;
         peertest::scenarios::basic::test_history_find_nodes_zero_distance(&target, &peertest).await;

trin-history/src/jsonrpc.rs

@@ -57,17 +57,21 @@ async fn complete_request(network: Arc<RwLock<HistoryNetwork>>, request: History
         HistoryEndpoint::TraceRecursiveFindContent(content_key) => {
             recursive_find_content(network, content_key, true).await
         }
+        HistoryEndpoint::AddEnr(enr) => add_enr(network, enr).await,
         HistoryEndpoint::DataRadius => {
             let radius = network.read().await.overlay.data_radius();
             Ok(json!(*radius))
         }
+        HistoryEndpoint::DeleteEnr(node_id) => delete_enr(network, node_id).await,
         HistoryEndpoint::FindContent(enr, content_key) => {
             find_content(network, enr, content_key).await
         }
         HistoryEndpoint::FindNodes(enr, distances) => find_nodes(network, enr, distances).await,
+        HistoryEndpoint::GetEnr(node_id) => get_enr(network, node_id).await,
         HistoryEndpoint::Gossip(content_key, content_value) => {
             gossip(network, content_key, content_value).await
         }
+        HistoryEndpoint::LookupEnr(node_id) => lookup_enr(network, node_id).await,
         HistoryEndpoint::Offer(enr, content_key, content_value) => {
             offer(network, enr, content_key, content_value).await
         }
@@ -188,11 +192,60 @@ async fn store(
         .put::<HistoryContentKey, Vec<u8>>(content_key, data)
     {
         Ok(_) => Ok(Value::Bool(true)),
-        Err(msg) => Ok(Value::String(msg.to_string())),
+        Err(err) => Ok(Value::String(err.to_string())),
     };
     response
 }
 
+/// Constructs a JSON call for the AddEnr method.
+async fn add_enr(
+    network: Arc<RwLock<HistoryNetwork>>,
+    enr: discv5::enr::Enr<discv5::enr::CombinedKey>,
+) -> Result<Value, String> {
+    let overlay = network.read().await.overlay.clone();
+    match overlay.add_enr(enr) {
+        Ok(_) => Ok(json!(true)),
+        Err(err) => Err(format!("AddEnr failed: {err:?}")),
+    }
+}
+
+/// Constructs a JSON call for the GetEnr method.
+async fn get_enr(
+    network: Arc<RwLock<HistoryNetwork>>,
+    node_id: ethportal_api::NodeId,
+) -> Result<Value, String> {
+    let node_id = discv5::enr::NodeId::from(node_id.0);
+    let overlay = network.read().await.overlay.clone();
+    match overlay.get_enr(node_id) {
+        Ok(enr) => Ok(json!(enr)),
+        Err(err) => Err(format!("GetEnr failed: {err:?}")),
+    }
+}
+
+/// Constructs a JSON call for the deleteEnr method.
+async fn delete_enr(
+    network: Arc<RwLock<HistoryNetwork>>,
+    node_id: ethportal_api::NodeId,
+) -> Result<Value, String> {
+    let node_id = discv5::enr::NodeId::from(node_id.0);
+    let overlay = network.read().await.overlay.clone();
+    let is_deleted = overlay.delete_enr(node_id);
+    Ok(json!(is_deleted))
+}
+
+/// Constructs a JSON call for the LookupEnr method.
+async fn lookup_enr(
+    network: Arc<RwLock<HistoryNetwork>>,
+    node_id: ethportal_api::NodeId,
+) -> Result<Value, String> {
+    let node_id = discv5::enr::NodeId::from(node_id.0);
+    let overlay = network.read().await.overlay.clone();
+    match overlay.lookup_enr(node_id).await {
+        Ok(enr) => Ok(json!(enr)),
+        Err(err) => Err(format!("LookupEnr failed: {err:?}")),
+    }
+}
+
 /// Constructs a JSON call for the FindContent method.
 async fn find_content(
     network: Arc<RwLock<HistoryNetwork>>,

trin-types/src/jsonrpc/endpoints.rs

@@ -27,14 +27,22 @@ pub enum StateEndpoint {
 /// History network JSON-RPC endpoints. Start with "portal_history" prefix
 #[derive(Debug, PartialEq, Eq, Clone)]
 pub enum HistoryEndpoint {
+    /// params: [enr]
+    AddEnr(Enr),
     /// params: None
     DataRadius,
+    /// params: [node_id]
+    DeleteEnr(NodeId),
     /// params: [enr, content_key]
     FindContent(Enr, HistoryContentKey),
     /// params: [enr, distances]
     FindNodes(Enr, Vec<u16>),
+    /// params: [node_id]
+    GetEnr(NodeId),
     /// params: content_key
     LocalContent(HistoryContentKey),
+    /// params: [node_id]
+    LookupEnr(NodeId),
     /// params: [content_key, content_value]
     Gossip(HistoryContentKey, HistoryContentValue),
     /// params: [enr, content_key]