Rename HeaderOracle and use infura mock for testing

Author: njgheorghita

Cargo.lock

@@ -13,7 +13,7 @@ repository = "https://github.com/ethereum/trin"
 discv5 = { git = "https://github.com/sigp/discv5.git", branch = "master" }
 ethereum-types = "0.12.1"
 hex = "0.4.3"
-log = "0.4.14"
+log = "0.4.17"
 prometheus_exporter = "0.8.4"
 rand = "0.8.4"
 rlp = "0.5.0"

Cargo.toml

@@ -7,7 +7,7 @@ rust-version = "1.58.0"
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
-anyhow = "1.0.52"
+anyhow = "1.0.57"
 clap = "2.33.3"
 discv5 = { git = "https://github.com/sigp/discv5.git", branch = "master" }
 futures = "0.3.21"

ethportal-peertest/Cargo.toml

@@ -362,7 +362,10 @@ pub fn make_ipc_request(ipc_path: &str, request: &JsonRpcRequest) -> anyhow::Res
     stream.flush().unwrap();
     let deser = serde_json::Deserializer::from_reader(stream);
     let next_obj = deser.into_iter::<Value>().next();
-    let response_obj = next_obj.ok_or(anyhow!("Empty JsonRpc response"))?;
+    let response_obj = match next_obj {
+        Some(val) => val,
+        None => return Err(anyhow!("Empty JsonRpc response")),
+    };
     get_response_result(response_obj)
 }
 

ethportal-peertest/src/jsonrpc.rs

@@ -42,7 +42,7 @@ pub async fn launch_node(id: u16, bootnode_enr: Option<&SszEnr>) -> anyhow::Resu
         Some(enr) => {
             let external_addr = format!(
                 "{}:{}",
-                enr.ip().expect("bootnode must have IP"),
+                enr.ip4().expect("bootnode must have IP"),
                 discovery_port
             );
             let enr_base64 = enr.to_base64();

ethportal-peertest/src/lib.rs

@@ -14,31 +14,27 @@ use trin_core::{
         discovery::Discovery, events::PortalnetEvents, storage::PortalStorage,
         types::messages::PortalnetConfig,
     },
-    types::validation::ValidationOracle,
+    types::validation::HeaderOracle,
     utils::bootnodes::parse_bootnodes,
-    utils::infura::get_infura_url,
     utp::stream::UtpListener,
 };
 use trin_history::initialize_history_network;
 use trin_state::initialize_state_network;
 
 pub async fn run_trin(
     trin_config: TrinConfig,
-    infura_project_id: String,
+    infura_url: String,
 ) -> Result<Arc<JsonRpcExiter>, Box<dyn std::error::Error>> {
     trin_config.display_config();
 
     let bootnode_enrs = parse_bootnodes(&trin_config.bootnodes)?;
-    let infura_url = get_infura_url(&infura_project_id);
-
     let portalnet_config = PortalnetConfig {
         external_addr: trin_config.external_addr,
         private_key: trin_config.private_key.clone(),
         listen_port: trin_config.discovery_port,
         no_stun: trin_config.no_stun,
         enable_metrics: trin_config.enable_metrics,
         bootnode_enrs,
-        infura_url: infura_url.clone(),
         ..Default::default()
     };
 
@@ -65,53 +61,53 @@ pub async fn run_trin(
         PortalStorage::setup_config(discovery.local_enr().node_id(), trin_config.kb)?;
 
     // Initialize validation oracle
-    let validation_oracle = ValidationOracle {
-        infura_url: portalnet_config.infura_url.clone(),
-        ..ValidationOracle::default()
+    let header_oracle = HeaderOracle {
+        infura_url: infura_url.clone(),
+        ..HeaderOracle::default()
     };
 
     debug!("Selected networks to spawn: {:?}", trin_config.networks);
+    // Initialize state sub-network service and event handlers, if selected
+    let (state_handler, state_network_task, state_event_tx, state_jsonrpc_tx) =
+        if trin_config.networks.iter().any(|val| val == STATE_NETWORK) {
+            initialize_state_network(
+                &discovery,
+                utp_listener_tx.clone(),
+                portalnet_config.clone(),
+                storage_config.clone(),
+                // todo: pass header_oracle into state network
+                // - to initialize StateValidator
+                // - for oracle to pick up handle to state_jsonrpc_tx
+            )
+            .await
+        } else {
+            (None, None, None, None)
+        };
+
     // Initialize chain history sub-network service and event handlers, if selected
     let (
         history_handler,
         history_network_task,
         history_event_tx,
         history_jsonrpc_tx,
-        _validation_oracle,
+        _header_oracle,
     ) = if trin_config
         .networks
         .iter()
         .any(|val| val == HISTORY_NETWORK)
     {
         initialize_history_network(
             &discovery,
-            utp_listener_tx.clone(),
+            utp_listener_tx,
             portalnet_config.clone(),
             storage_config.clone(),
-            validation_oracle.clone(),
+            header_oracle.clone(),
         )
         .await
     } else {
         (None, None, None, None, None)
     };
 
-    // Initialize state sub-network service and event handlers, if selected
-    let (state_handler, state_network_task, state_event_tx, state_jsonrpc_tx) =
-        if trin_config.networks.iter().any(|val| val == STATE_NETWORK) {
-            initialize_state_network(
-                &discovery,
-                utp_listener_tx,
-                portalnet_config.clone(),
-                storage_config.clone(),
-                // todo: pass validation_oracle into state network
-                // - to initialize StateValidator
-                // - for oracle to pick up handle to state_jsonrpc_tx
-            )
-            .await
-        } else {
-            (None, None, None, None)
-        };
-
     // Initialize json-rpc server
     let (portal_jsonrpc_tx, portal_jsonrpc_rx) = mpsc::unbounded_channel::<PortalJsonRpcRequest>();
     let jsonrpc_trin_config = trin_config.clone();

src/lib.rs

@@ -1,5 +1,5 @@
 use trin_core::cli::TrinConfig;
-use trin_core::utils::infura::fetch_infura_id_from_env;
+use trin_core::utils::infura::build_infura_project_url_from_env;
 
 use trin::run_trin;
 
@@ -10,9 +10,9 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     println!("Launching trin");
 
     let trin_config = TrinConfig::from_cli();
-    let infura_project_id = fetch_infura_id_from_env();
+    let infura_url = build_infura_project_url_from_env();
 
-    let exiter = run_trin(trin_config, infura_project_id).await?;
+    let exiter = run_trin(trin_config, infura_url).await?;
 
     tokio::signal::ctrl_c()
         .await

src/main.rs

@@ -5,7 +5,7 @@ edition = "2021"
 rust-version = "1.58.0"
 
 [dependencies]
-anyhow = "1.0.52"
+anyhow = "1.0.57"
 async-recursion = "1.0.0"
 async-trait = "0.1.53"
 base64 = "0.13.0"
@@ -27,7 +27,7 @@ hmac-sha256 = "1.1.1"
 httparse = "1.5.1"
 keccak-hash = "0.8.0"
 lazy_static = "1.4.0"
-log = "0.4.14"
+log = "0.4.17"
 num = "0.4.0"
 parking_lot = "0.11.2"
 prometheus_exporter = "0.8.4"

trin-core/Cargo.toml

@@ -91,7 +91,7 @@ impl Discovery {
             if let Some(ip_address) = config.enr_address {
                 builder.ip(ip_address);
             }
-            builder.udp(config.listen_port);
+            builder.udp4(config.listen_port);
             builder.build(&enr_key).unwrap()
         };
 
@@ -137,7 +137,7 @@ impl Discovery {
         json!({
             "enr":  self.discv5.local_enr().to_base64(),
             "nodeId":  self.discv5.local_enr().node_id().to_string(),
-            "ip":  self.discv5.local_enr().ip().map_or("None".to_owned(), |ip| ip.to_string())
+            "ip":  self.discv5.local_enr().ip4().map_or("None".to_owned(), |ip| ip.to_string())
         })
     }
 

trin-core/src/portalnet/discovery.rs

@@ -1,5 +1,10 @@
 use anyhow::anyhow;
-use std::{collections::HashSet, marker::PhantomData, sync::Arc, time::Duration};
+use std::{
+    collections::HashSet,
+    marker::{PhantomData, Sync},
+    sync::Arc,
+    time::Duration,
+};
 
 use super::{
     discovery::Discovery,
@@ -92,7 +97,7 @@ pub struct OverlayProtocol<TContentKey, TMetric, TValidator> {
 }
 
 impl<
-        TContentKey: OverlayContentKey + Send,
+        TContentKey: OverlayContentKey + Send + Sync,
         TMetric: Metric + Send,
         TValidator: 'static + Validator<TContentKey> + Send,
     > OverlayProtocol<TContentKey, TMetric, TValidator>

trin-core/src/portalnet/overlay.rs

@@ -1,4 +1,11 @@
-use std::{collections::HashMap, fmt, marker::PhantomData, sync::Arc, task::Poll, time::Duration};
+use std::{
+    collections::HashMap,
+    fmt,
+    marker::{PhantomData, Sync},
+    sync::Arc,
+    task::Poll,
+    time::Duration,
+};
 
 use crate::{
     portalnet::{
@@ -266,7 +273,7 @@ pub struct OverlayService<TContentKey, TMetric, TValidator> {
 }
 
 impl<
-        TContentKey: OverlayContentKey + Send,
+        TContentKey: OverlayContentKey + Send + Sync,
         TMetric: Metric + Send,
         TValidator: 'static + Validator<TContentKey> + Send,
     > OverlayService<TContentKey, TMetric, TValidator>
@@ -966,18 +973,23 @@ impl<
                 return;
             }
         };
-        let ck_clone = content_key.clone();
-        let content_clone = content.clone();
-        self.validator
-            .validate_content(ck_clone, content_clone)
-            .await;
+        match self
+            .validator
+            .validate_content(&content_key, &content)
+            .await
+        {
+            Ok(_) => (),
+            Err(_) => {
+                error!("Unable to validate received content.");
+                return;
+            }
+        }
         match self.storage.read().should_store(&content_key) {
             Ok(should_store) => match should_store {
-                true => self
-                    .storage
-                    .write()
-                    .store(&content_key, &content.into())
-                    .unwrap(),
+                true => match self.storage.write().store(&content_key, &content.into()) {
+                    Ok(_) => (),
+                    Err(_) => error!("Content received, but not stored: Error writing to db."),
+                },
                 false => debug!(
                     "Content received, but not stored: Distance falls outside current radius."
                 ),
@@ -1285,7 +1297,7 @@ mod tests {
                 content_key::IdentityContentKey, messages::PortalnetConfig, metric::XorMetric,
             },
         },
-        types::validation::{IdentityValidator, ValidationOracle},
+        types::validation::{HeaderOracle, IdentityValidator},
         utils::node_id::generate_random_remote_enr,
     };
 
@@ -1331,8 +1343,8 @@ mod tests {
         let (request_tx, request_rx) = mpsc::unbounded_channel();
         let (response_tx, response_rx) = mpsc::unbounded_channel();
         let metrics = None;
-        let validation_oracle = ValidationOracle::default();
-        let validator = IdentityValidator { validation_oracle };
+        let header_oracle = HeaderOracle::default();
+        let validator = IdentityValidator { header_oracle };
 
         OverlayService {
             discovery,
@@ -1657,7 +1669,7 @@ mod tests {
 
         // Modify first ENR to increment sequence number.
         let updated_udp: u16 = 9000;
-        let _ = enr1.set_udp(updated_udp, &sk1);
+        let _ = enr1.set_udp4(updated_udp, &sk1);
         assert_ne!(1, enr1.seq());
 
         let mut enrs: Vec<Enr> = vec![];

trin-core/src/portalnet/overlay_service.rs

@@ -117,7 +117,6 @@ pub struct PortalnetConfig {
     pub data_radius: U256,
     pub no_stun: bool,
     pub enable_metrics: bool,
-    pub infura_url: String,
 }
 
 impl Default for PortalnetConfig {
@@ -130,7 +129,6 @@ impl Default for PortalnetConfig {
             data_radius: U256::from(u64::MAX), //TODO better data_radius default?
             no_stun: false,
             enable_metrics: false,
-            infura_url: "https://mainnet.infura.io:443/v3/".to_string(),
         }
     }
 }

trin-core/src/portalnet/types/messages.rs

@@ -7,12 +7,11 @@ use crate::jsonrpc::types::{HistoryJsonRpcRequest, JsonRequest, Params};
 use crate::portalnet::types::content_key::IdentityContentKey;
 use crate::portalnet::types::messages::ByteList;
 
-// aka. HeaderOracle
 // This struct is responsible for dispatching cross-overlay-network requests
 // for data to perform validation. Currently, it just proxies these requests
 // on to infura.
 #[derive(Clone)]
-pub struct ValidationOracle {
+pub struct HeaderOracle {
     pub infura_url: String,
     // We could simply store the main portal jsonrpc tx channel here, rather than each
     // individual channel. But my sense is that this will be more useful in terms of
@@ -23,10 +22,10 @@ pub struct ValidationOracle {
     pub state_jsonrpc_tx: Option<bool>,
 }
 
-impl Default for ValidationOracle {
+impl Default for HeaderOracle {
     fn default() -> Self {
         Self {
-            infura_url: "https://mainnet.infura.io::443/v3/".to_string(),
+            infura_url: "https://mainnet.infura.io:443/v3/".to_string(),
             history_jsonrpc_tx: None,
             header_gossip_jsonrpc_tx: None,
             block_indices_jsonrpc_tx: None,
@@ -35,7 +34,7 @@ impl Default for ValidationOracle {
     }
 }
 
-impl ValidationOracle {
+impl HeaderOracle {
     // Currently falls back to infura, to be updated to use canonical block indices network.
     pub fn get_hash_at_height(&mut self, hex_number: String) -> anyhow::Result<String> {
         let request = JsonRequest {
@@ -73,21 +72,30 @@ impl ValidationOracle {
 // This trait is used by all overlay-network Validators to validate content in the overlay service.
 #[async_trait]
 pub trait Validator<TContentKey> {
-    async fn validate_content(&mut self, content_key: TContentKey, content: ByteList)
+    async fn validate_content(
+        &mut self,
+        content_key: &TContentKey,
+        content: &ByteList,
+    ) -> anyhow::Result<()>
     where
         TContentKey: 'async_trait;
 }
 
 // This is a mock Validator for use in tests where no validation is required.
 pub struct IdentityValidator {
-    pub validation_oracle: ValidationOracle,
+    pub header_oracle: HeaderOracle,
 }
 
 #[async_trait]
 impl Validator<IdentityContentKey> for IdentityValidator {
-    async fn validate_content(&mut self, _content_key: IdentityContentKey, _content: ByteList)
+    async fn validate_content(
+        &mut self,
+        _content_key: &IdentityContentKey,
+        _content: &ByteList,
+    ) -> anyhow::Result<()>
     where
         IdentityContentKey: 'async_trait,
     {
+        Ok(())
     }
 }