Write initial ValidationOracle implementation.

Author: njgheorghita

Cargo.lock

@@ -14,7 +14,9 @@ use trin_core::{
         discovery::Discovery, events::PortalnetEvents, storage::PortalStorage,
         types::messages::PortalnetConfig,
     },
+    types::validation::ValidationOracle,
     utils::bootnodes::parse_bootnodes,
+    utils::infura::get_infura_url,
     utp::stream::UtpListener,
 };
 use trin_history::initialize_history_network;
@@ -27,6 +29,7 @@ pub async fn run_trin(
     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,
@@ -35,6 +38,7 @@ pub async fn run_trin(
         no_stun: trin_config.no_stun,
         enable_metrics: trin_config.enable_metrics,
         bootnode_enrs,
+        infura_url: infura_url.clone(),
         ..Default::default()
     };
 
@@ -60,33 +64,48 @@ pub async fn run_trin(
     let storage_config =
         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()
+    };
+
     debug!("Selected networks to spawn: {:?}", trin_config.networks);
+    // 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,
+    ) = if trin_config
+        .networks
+        .iter()
+        .any(|val| val == HISTORY_NETWORK)
+    {
+        initialize_history_network(
+            &discovery,
+            utp_listener_tx.clone(),
+            portalnet_config.clone(),
+            storage_config.clone(),
+            validation_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.clone(),
-                portalnet_config.clone(),
-                storage_config.clone(),
-            )
-            .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) =
-        if trin_config
-            .networks
-            .iter()
-            .any(|val| val == HISTORY_NETWORK)
-        {
-            initialize_history_network(
                 &discovery,
                 utp_listener_tx,
                 portalnet_config.clone(),
                 storage_config.clone(),
+                // todo: validation oracle gets passed into state network
+                // - for StateValidator to use
+                // - for oracle to pick up handle to state_jsonrpc_tx
             )
             .await
         } else {
@@ -103,7 +122,7 @@ pub async fn run_trin(
         tokio::task::spawn_blocking(|| {
             launch_jsonrpc_server(
                 jsonrpc_trin_config,
-                infura_project_id,
+                infura_url,
                 portal_jsonrpc_tx,
                 live_server_tx,
                 json_exiter_clone,

src/lib.rs

@@ -1,6 +1,5 @@
-use std::env;
-
 use trin_core::cli::TrinConfig;
+use trin_core::utils::infura::fetch_infura_id_from_env;
 
 use trin::run_trin;
 
@@ -11,13 +10,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     println!("Launching trin");
 
     let trin_config = TrinConfig::from_cli();
-    let infura_project_id = match env::var("TRIN_INFURA_PROJECT_ID") {
-        Ok(val) => val,
-        Err(_) => panic!(
-            "Must supply Infura key as environment variable, like:\n\
-            TRIN_INFURA_PROJECT_ID=\"your-key-here\" trin"
-        ),
-    };
+    let infura_project_id = fetch_infura_id_from_env();
 
     let exiter = run_trin(trin_config, infura_project_id).await?;
 

src/main.rs

@@ -7,6 +7,7 @@ rust-version = "1.58.0"
 [dependencies]
 anyhow = "1.0.52"
 async-recursion = "1.0.0"
+async-trait = "0.1.53"
 base64 = "0.13.0"
 bytes = "1.1.0"
 clap = "2.33.3"

trin-core/Cargo.toml

@@ -58,7 +58,7 @@ impl Default for JsonRpcExiter {
 
 pub fn launch_jsonrpc_server(
     trin_config: TrinConfig,
-    infura_project_id: String,
+    infura_url: String,
     portal_tx: UnboundedSender<PortalJsonRpcRequest>,
     live_server_tx: tokio::sync::mpsc::Sender<bool>,
     json_rpc_exiter: Arc<JsonRpcExiter>,
@@ -68,19 +68,13 @@ pub fn launch_jsonrpc_server(
     match trin_config.web3_transport.as_str() {
         "ipc" => launch_ipc_client(
             pool,
-            infura_project_id,
+            infura_url,
             &trin_config.web3_ipc_path,
             portal_tx,
             live_server_tx,
             json_rpc_exiter,
         ),
-        "http" => launch_http_client(
-            pool,
-            infura_project_id,
-            trin_config,
-            portal_tx,
-            live_server_tx,
-        ),
+        "http" => launch_http_client(pool, infura_url, trin_config, portal_tx, live_server_tx),
         val => panic!("Unsupported web3 transport: {}", val),
     }
 }
@@ -126,7 +120,7 @@ fn get_listener_result(ipc_path: &str) -> tokio::io::Result<uds_windows::UnixLis
 
 fn launch_ipc_client(
     pool: ThreadPool,
-    infura_project_id: String,
+    infura_url: String,
     ipc_path: &str,
     portal_tx: UnboundedSender<PortalJsonRpcRequest>,
     live_server_tx: tokio::sync::mpsc::Sender<bool>,
@@ -174,10 +168,9 @@ fn launch_ipc_client(
             Err(_) => break, // Socket exited
         };
         debug!("New IPC client: {:?}", stream.peer_addr().unwrap());
-        let infura_project_id = infura_project_id.clone();
+        let infura_url = infura_url.clone();
         let portal_tx = portal_tx.clone();
         pool.execute(move || {
-            let infura_url = get_infura_url(&infura_project_id);
             let mut rx = stream.try_clone().unwrap();
             let mut tx = stream;
             serve_ipc_client(&mut rx, &mut tx, &infura_url, portal_tx);
@@ -192,7 +185,7 @@ fn launch_ipc_client(
 
 fn launch_http_client(
     pool: ThreadPool,
-    infura_project_id: String,
+    infura_url: String,
     trin_config: TrinConfig,
     portal_tx: UnboundedSender<PortalJsonRpcRequest>,
     live_server_tx: tokio::sync::mpsc::Sender<bool>,
@@ -212,10 +205,9 @@ fn launch_http_client(
     for stream in listener.incoming() {
         match stream {
             Ok(stream) => {
-                let infura_project_id = infura_project_id.clone();
+                let infura_url = infura_url.clone();
                 let portal_tx = portal_tx.clone();
                 pool.execute(move || {
-                    let infura_url = get_infura_url(&infura_project_id);
                     serve_http_client(stream, &infura_url, portal_tx);
                 });
             }
@@ -386,7 +378,7 @@ fn dispatch_trin_request(
 }
 
 // Handle all requests served by infura
-fn dispatch_infura_request(obj: JsonRequest, infura_url: &str) -> Result<String, String> {
+pub fn dispatch_infura_request(obj: JsonRequest, infura_url: &str) -> Result<String, String> {
     match proxy_to_url(&obj, infura_url) {
         Ok(result_body) => Ok(std::str::from_utf8(&result_body).unwrap().to_owned()),
         Err(err) => Err(json!({
@@ -447,7 +439,3 @@ fn proxy_to_url(request: &JsonRequest, url: &str) -> io::Result<Vec<u8>> {
         )),
     }
 }
-
-fn get_infura_url(infura_project_id: &str) -> String {
-    return format!("https://mainnet.infura.io:443/v3/{}", infura_project_id);
-}

trin-core/src/jsonrpc/service.rs

@@ -15,6 +15,7 @@ use crate::portalnet::{
     },
 };
 
+use crate::types::validation::Validator;
 use crate::utp::{
     stream::{UtpListenerRequest, UtpSocket, BUF_SIZE},
     trin_helpers::{UtpAccept, UtpMessage},
@@ -65,7 +66,7 @@ impl Default for OverlayConfig {
 /// implement the overlay protocol and the overlay protocol is where we can encapsulate the logic for
 /// handling common network requests/responses.
 #[derive(Clone)]
-pub struct OverlayProtocol<TContentKey, TMetric> {
+pub struct OverlayProtocol<TContentKey, TMetric, TValidator> {
     /// Reference to the underlying discv5 protocol
     pub discovery: Arc<Discovery>,
     /// Reference to the database instance
@@ -86,10 +87,15 @@ pub struct OverlayProtocol<TContentKey, TMetric> {
     phantom_content_key: PhantomData<TContentKey>,
     /// Associate a metric with the overlay network.
     phantom_metric: PhantomData<TMetric>,
+    /// Declare the Validator type for a given overlay network.
+    phantom_validator: PhantomData<TValidator>,
 }
 
-impl<TContentKey: OverlayContentKey + Send, TMetric: Metric + Send>
-    OverlayProtocol<TContentKey, TMetric>
+impl<
+        TContentKey: OverlayContentKey + Send,
+        TMetric: Metric + Send,
+        TValidator: 'static + Validator<TContentKey> + Send,
+    > OverlayProtocol<TContentKey, TMetric, TValidator>
 {
     pub async fn new(
         config: OverlayConfig,
@@ -98,6 +104,7 @@ impl<TContentKey: OverlayContentKey + Send, TMetric: Metric + Send>
         storage: Arc<RwLock<PortalStorage>>,
         data_radius: U256,
         protocol: ProtocolId,
+        validator: TValidator,
     ) -> Self {
         let kbuckets = Arc::new(RwLock::new(KBucketsTable::new(
             discovery.local_enr().node_id().into(),
@@ -108,7 +115,7 @@ impl<TContentKey: OverlayContentKey + Send, TMetric: Metric + Send>
         )));
 
         let data_radius = Arc::new(data_radius);
-        let request_tx = OverlayService::<TContentKey, TMetric>::spawn(
+        let request_tx = OverlayService::<TContentKey, TMetric, TValidator>::spawn(
             Arc::clone(&discovery),
             Arc::clone(&storage),
             Arc::clone(&kbuckets),
@@ -118,6 +125,7 @@ impl<TContentKey: OverlayContentKey + Send, TMetric: Metric + Send>
             protocol.clone(),
             utp_listener_tx.clone(),
             config.enable_metrics,
+            validator,
         )
         .await
         .unwrap();
@@ -132,6 +140,7 @@ impl<TContentKey: OverlayContentKey + Send, TMetric: Metric + Send>
             utp_listener_tx,
             phantom_content_key: PhantomData,
             phantom_metric: PhantomData,
+            phantom_validator: PhantomData,
         }
     }