Wrap HeaderOracle in arc/rwlock

njgheorghita · njgheorghita · commit 17dd835873e4 · 2022-06-08T13:46:55.000+02:00
diff --git a/src/lib.rs b/src/lib.rs
@@ -1,4 +1,4 @@
-use std::sync::Arc;
+use std::sync::{Arc, RwLock};
 
 use log::debug;
 use tokio::sync::mpsc;
@@ -61,10 +61,10 @@ pub async fn run_trin(
         PortalStorage::setup_config(discovery.local_enr().node_id(), trin_config.kb)?;
 
     // Initialize validation oracle
-    let header_oracle = HeaderOracle {
+    let header_oracle = Arc::new(RwLock::new(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
@@ -85,28 +85,23 @@ pub async fn run_trin(
         };
 
     // Initialize chain history sub-network service and event handlers, if selected
-    let (
-        history_handler,
-        history_network_task,
-        history_event_tx,
-        history_jsonrpc_tx,
-        _header_oracle,
-    ) = if trin_config
-        .networks
-        .iter()
-        .any(|val| val == HISTORY_NETWORK)
-    {
-        initialize_history_network(
-            &discovery,
-            utp_listener_tx,
-            portalnet_config.clone(),
-            storage_config.clone(),
-            header_oracle.clone(),
-        )
-        .await
-    } else {
-        (None, None, None, None, None)
-    };
+    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(),
+                header_oracle,
+            )
+            .await
+        } else {
+            (None, None, None, None)
+        };
 
     // Initialize json-rpc server
     let (portal_jsonrpc_tx, portal_jsonrpc_rx) = mpsc::unbounded_channel::<PortalJsonRpcRequest>();
diff --git a/trin-core/src/portalnet/overlay_service.rs b/trin-core/src/portalnet/overlay_service.rs
@@ -1020,10 +1020,10 @@ impl<
             Ok(should_store) => match should_store {
                 true => match self.storage.write().store(&content_key, &content.into()) {
                     Ok(_) => (),
-                    Err(_) => error!("Content received, but not stored: Error writing to db."),
+                    Err(msg) => error!("{}", format!("Content received, but not stored: {msg}")),
                 },
                 false => debug!(
-                    "Content received, but not stored: Distance falls outside current radius."
+                    "Content received, but not stored: Content is already stored or its distance falls outside current radius."
                 ),
             },
             Err(_) => {
@@ -1329,7 +1329,7 @@ mod tests {
                 content_key::IdentityContentKey, messages::PortalnetConfig, metric::XorMetric,
             },
         },
-        types::validation::{HeaderOracle, IdentityValidator},
+        types::validation::IdentityValidator,
         utils::node_id::generate_random_remote_enr,
     };
 
@@ -1375,8 +1375,7 @@ mod tests {
         let (request_tx, request_rx) = mpsc::unbounded_channel();
         let (response_tx, response_rx) = mpsc::unbounded_channel();
         let metrics = None;
-        let header_oracle = HeaderOracle::default();
-        let validator = IdentityValidator { header_oracle };
+        let validator = IdentityValidator {};
 
         OverlayService {
             discovery,
diff --git a/trin-core/src/types/validation.rs b/trin-core/src/types/validation.rs
@@ -82,9 +82,7 @@ pub trait Validator<TContentKey> {
 }
 
 // This is a mock Validator for use in tests where no validation is required.
-pub struct IdentityValidator {
-    pub header_oracle: HeaderOracle,
-}
+pub struct IdentityValidator {}
 
 #[async_trait]
 impl Validator<IdentityContentKey> for IdentityValidator {
diff --git a/trin-core/tests/overlay.rs b/trin-core/tests/overlay.rs
@@ -14,7 +14,7 @@ use trin_core::{
         Enr,
     },
     socket,
-    types::validation::{HeaderOracle, IdentityValidator},
+    types::validation::IdentityValidator,
 };
 
 use discv5::Discv5Event;
@@ -40,8 +40,7 @@ async fn init_overlay(
     let overlay_config = OverlayConfig::default();
     // Ignore all uTP events
     let (utp_listener_tx, _) = unbounded_channel::<UtpListenerRequest>();
-    let header_oracle = HeaderOracle::default();
-    let validator = IdentityValidator { header_oracle };
+    let validator = IdentityValidator {};
 
     OverlayProtocol::new(
         overlay_config,
diff --git a/trin-history/src/lib.rs b/trin-history/src/lib.rs
@@ -3,14 +3,15 @@ mod jsonrpc;
 pub mod network;
 pub mod validation;
 
-use log::info;
-use tokio::{sync::mpsc, task::JoinHandle};
+use std::sync::{Arc, RwLock};
 
-use crate::{events::HistoryEvents, jsonrpc::HistoryRequestHandler};
 use discv5::TalkRequest;
+use log::info;
 use network::HistoryNetwork;
-use std::sync::Arc;
 use tokio::sync::mpsc::UnboundedSender;
+use tokio::{sync::mpsc, task::JoinHandle};
+
+use crate::{events::HistoryEvents, jsonrpc::HistoryRequestHandler};
 use trin_core::{
     cli::TrinConfig,
     jsonrpc::types::HistoryJsonRpcRequest,
@@ -76,10 +77,10 @@ pub async fn main(infura_url: String) -> Result<(), Box<dyn std::error::Error>>
         events.process_discv5_requests().await;
     });
 
-    let header_oracle = HeaderOracle {
+    let header_oracle = Arc::new(RwLock::new(HeaderOracle {
         infura_url,
         ..HeaderOracle::default()
-    };
+    }));
     let history_network = HistoryNetwork::new(
         discovery.clone(),
         overlay_sender,
@@ -100,31 +101,29 @@ type HistoryHandler = Option<HistoryRequestHandler>;
 type HistoryNetworkTask = Option<JoinHandle<()>>;
 type HistoryEventTx = Option<mpsc::UnboundedSender<TalkRequest>>;
 type HistoryJsonRpcTx = Option<mpsc::UnboundedSender<HistoryJsonRpcRequest>>;
-type HistoryHeaderOracle = Option<HeaderOracle>;
 
 pub async fn initialize_history_network(
     discovery: &Arc<Discovery>,
     utp_listener_tx: UnboundedSender<UtpListenerRequest>,
     portalnet_config: PortalnetConfig,
     storage_config: PortalStorageConfig,
-    mut header_oracle: HeaderOracle,
+    header_oracle: Arc<RwLock<HeaderOracle>>,
 ) -> (
     HistoryHandler,
     HistoryNetworkTask,
     HistoryEventTx,
     HistoryJsonRpcTx,
-    HistoryHeaderOracle,
 ) {
     let (history_jsonrpc_tx, history_jsonrpc_rx) =
         mpsc::unbounded_channel::<HistoryJsonRpcRequest>();
-    header_oracle.history_jsonrpc_tx = Some(history_jsonrpc_tx.clone());
+    header_oracle.write().unwrap().history_jsonrpc_tx = Some(history_jsonrpc_tx.clone());
     let (history_event_tx, history_event_rx) = mpsc::unbounded_channel::<TalkRequest>();
     let history_network = HistoryNetwork::new(
         Arc::clone(discovery),
         utp_listener_tx,
         storage_config,
         portalnet_config.clone(),
-        header_oracle.clone(),
+        header_oracle,
     )
     .await;
     let history_network = Arc::new(history_network);
@@ -142,7 +141,6 @@ pub async fn initialize_history_network(
         Some(history_network_task),
         Some(history_event_tx),
         Some(history_jsonrpc_tx),
-        Some(header_oracle),
     )
 }
 
diff --git a/trin-history/src/network.rs b/trin-history/src/network.rs
@@ -1,6 +1,7 @@
 use anyhow::anyhow;
 use log::debug;
 use std::sync::Arc;
+use std::sync::RwLock as StdRwLock;
 
 use parking_lot::RwLock;
 use tokio::sync::mpsc::UnboundedSender;
@@ -34,7 +35,7 @@ impl HistoryNetwork {
         utp_listener_tx: UnboundedSender<UtpListenerRequest>,
         storage_config: PortalStorageConfig,
         portal_config: PortalnetConfig,
-        header_oracle: HeaderOracle,
+        header_oracle: Arc<StdRwLock<HeaderOracle>>,
     ) -> Self {
         let config = OverlayConfig {
             bootnode_enrs: portal_config.bootnode_enrs.clone(),
diff --git a/trin-history/src/validation.rs b/trin-history/src/validation.rs
@@ -1,3 +1,5 @@
+use std::sync::{Arc, RwLock};
+
 use anyhow::anyhow;
 use async_trait::async_trait;
 use rlp::Rlp;
@@ -8,7 +10,7 @@ use trin_core::types::header::Header;
 use trin_core::types::validation::{HeaderOracle, Validator};
 
 pub struct ChainHistoryValidator {
-    pub header_oracle: HeaderOracle,
+    pub header_oracle: Arc<RwLock<HeaderOracle>>,
 }
 
 #[async_trait]
@@ -26,8 +28,13 @@ impl Validator<HistoryContentKey> for ChainHistoryValidator {
                 let rlp = Rlp::new(content);
                 let header = Header::decode_rlp(&rlp).expect("invalid header");
                 let number = format!("0x{:02X}", header.number);
-                let expected_hash = self.header_oracle.get_hash_at_height(number).unwrap();
-                let actual_hash = hex::encode(key.block_hash);
+                let expected_hash = &self
+                    .header_oracle
+                    .write()
+                    .unwrap()
+                    .get_hash_at_height(number)
+                    .unwrap();
+                let actual_hash = &hex::encode(key.block_hash);
                 if actual_hash == expected_hash {
                     Ok(())
                 } else {
@@ -120,10 +127,10 @@ mod tests {
 
         let header: Header = Header::decode_rlp(&rlp).expect("error decoding header");
         let infura_url = server.url("/get_header");
-        let header_oracle = HeaderOracle {
+        let header_oracle = Arc::new(RwLock::new(HeaderOracle {
             infura_url,
             ..HeaderOracle::default()
-        };
+        }));
         let mut chain_history_validator = ChainHistoryValidator { header_oracle };
         let content_key = HistoryContentKey::BlockHeader(BlockHeader {
             chain_id: 1,
@@ -149,10 +156,10 @@ mod tests {
         header.number = 669052;
 
         let infura_url = server.url("/get_header");
-        let header_oracle = HeaderOracle {
+        let header_oracle = Arc::new(RwLock::new(HeaderOracle {
             infura_url,
             ..HeaderOracle::default()
-        };
+        }));
         let mut chain_history_validator = ChainHistoryValidator { header_oracle };
         let content_key = HistoryContentKey::BlockHeader(BlockHeader {
             chain_id: 1,
@@ -179,10 +186,10 @@ mod tests {
         header.gas_limit = U256::from(3141591);
 
         let infura_url = server.url("/get_header");
-        let header_oracle = HeaderOracle {
+        let header_oracle = Arc::new(RwLock::new(HeaderOracle {
             infura_url,
             ..HeaderOracle::default()
-        };
+        }));
         let mut chain_history_validator = ChainHistoryValidator { header_oracle };
         let content_key = HistoryContentKey::BlockHeader(BlockHeader {
             chain_id: 1,

Original file line number	Diff line number	Diff line change
`@@ -82,9 +82,7 @@ pub trait Validator<TContentKey> {`
`82`	`82`	`}`
`83`	`83`
`84`	`84`	`// This is a mock Validator for use in tests where no validation is required.`
`85`		`-pub struct IdentityValidator {`
`86`		`- pub header_oracle: HeaderOracle,`
`87`		`-}`
	`85`	`+pub struct IdentityValidator {}`
`88`	`86`
`89`	`87`	`#[async_trait]`
`90`	`88`	`impl Validator<IdentityContentKey> for IdentityValidator {`