Move findcontent message processing up pipeline

Author: njgheorghita

trin-core/src/portalnet/overlay_service.rs

@@ -925,7 +925,17 @@ impl<
         match response {
             Response::Pong(pong) => self.process_pong(pong, source),
             Response::Nodes(nodes) => self.process_nodes(nodes, source),
-            Response::Content(content) => self.process_content(content, source, request).await,
+            Response::Content(content) => {
+                let find_content_request = match request {
+                    Request::FindContent(find_content) => find_content,
+                    _ => {
+                        error!("Unable to process received content: Invalid request message.");
+                        return;
+                    }
+                };
+                self.process_content(content, source, find_content_request)
+                    .await
+            }
             _ => {}
         }
     }
@@ -973,7 +983,7 @@ impl<
     }
 
     /// Processes a Content response.
-    async fn process_content(&mut self, content: Content, source: Enr, request: Request) {
+    async fn process_content(&mut self, content: Content, source: Enr, request: FindContent) {
         debug!(
             "[{:?}] Processing Content response from node. Node: {}",
             self.protocol,
@@ -990,15 +1000,8 @@ impl<
         }
     }
 
-    async fn process_received_content(&mut self, content: ByteList, request: Request) {
-        let find_content = match request {
-            Request::FindContent(find_content) => find_content,
-            _ => {
-                error!("Unable to process received content: Invalid request message.");
-                return;
-            }
-        };
-        let content_key = match TContentKey::try_from(find_content.content_key) {
+    async fn process_received_content(&mut self, content: ByteList, request: FindContent) {
+        let content_key = match TContentKey::try_from(request.content_key) {
             Ok(val) => val,
             Err(_) => {
                 error!("Unable to process received content: Invalid content key.");
@@ -1011,16 +1014,16 @@ impl<
             .await
         {
             Ok(_) => (),
-            Err(_) => {
-                error!("Unable to validate received content.");
+            Err(msg) => {
+                error!("Unable to validate received content: {msg:?}");
                 return;
             }
         }
         match self.storage.read().should_store(&content_key) {
             Ok(should_store) => match should_store {
                 true => match self.storage.write().store(&content_key, &content.into()) {
                     Ok(_) => (),
-                    Err(msg) => error!("{}", format!("Content received, but not stored: {msg}")),
+                    Err(msg) => error!("Content received, but not stored: {msg}"),
                 },
                 false => debug!(
                     "Content received, but not stored: Content is already stored or its distance falls outside current radius."
@@ -1329,7 +1332,7 @@ mod tests {
                 content_key::IdentityContentKey, messages::PortalnetConfig, metric::XorMetric,
             },
         },
-        types::validation::IdentityValidator,
+        types::validation::MockValidator,
         utils::node_id::generate_random_remote_enr,
     };
 
@@ -1344,7 +1347,7 @@ mod tests {
         };
     }
 
-    fn build_service() -> OverlayService<IdentityContentKey, XorMetric, IdentityValidator> {
+    fn build_service() -> OverlayService<IdentityContentKey, XorMetric, MockValidator> {
         let portal_config = PortalnetConfig {
             no_stun: true,
             ..Default::default()
@@ -1375,7 +1378,7 @@ mod tests {
         let (request_tx, request_rx) = mpsc::unbounded_channel();
         let (response_tx, response_rx) = mpsc::unbounded_channel();
         let metrics = None;
-        let validator = IdentityValidator {};
+        let validator = MockValidator {};
 
         OverlayService {
             discovery,

trin-core/src/types/validation.rs

@@ -7,9 +7,9 @@ use crate::jsonrpc::types::{HistoryJsonRpcRequest, JsonRequest, Params};
 use crate::portalnet::types::content_key::IdentityContentKey;
 use crate::portalnet::types::messages::ByteList;
 
-// This struct is responsible for dispatching cross-overlay-network requests
-// for data to perform validation. Currently, it just proxies these requests
-// on to infura.
+/// 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 HeaderOracle {
     pub infura_url: String,
@@ -69,7 +69,7 @@ impl HeaderOracle {
     }
 }
 
-// This trait is used by all overlay-network Validators to validate content in the overlay service.
+/// Used by all overlay-network Validators to validate content in the overlay service.
 #[async_trait]
 pub trait Validator<TContentKey> {
     async fn validate_content(
@@ -81,11 +81,11 @@ pub trait Validator<TContentKey> {
         TContentKey: 'async_trait;
 }
 
-// This is a mock Validator for use in tests where no validation is required.
-pub struct IdentityValidator {}
+/// For use in tests where no validation needs to be performed.
+pub struct MockValidator {}
 
 #[async_trait]
-impl Validator<IdentityContentKey> for IdentityValidator {
+impl Validator<IdentityContentKey> for MockValidator {
     async fn validate_content(
         &mut self,
         _content_key: &IdentityContentKey,

trin-core/tests/overlay.rs

@@ -14,7 +14,7 @@ use trin_core::{
         Enr,
     },
     socket,
-    types::validation::IdentityValidator,
+    types::validation::MockValidator,
 };
 
 use discv5::Discv5Event;
@@ -30,7 +30,7 @@ use trin_core::utp::stream::UtpListenerRequest;
 async fn init_overlay(
     discovery: Arc<Discovery>,
     protocol: ProtocolId,
-) -> OverlayProtocol<IdentityContentKey, XorMetric, IdentityValidator> {
+) -> OverlayProtocol<IdentityContentKey, XorMetric, MockValidator> {
     let storage_config = PortalStorage::setup_config(
         discovery.local_enr().node_id(),
         DEFAULT_STORAGE_CAPACITY.parse().unwrap(),
@@ -40,7 +40,7 @@ async fn init_overlay(
     let overlay_config = OverlayConfig::default();
     // Ignore all uTP events
     let (utp_listener_tx, _) = unbounded_channel::<UtpListenerRequest>();
-    let validator = IdentityValidator {};
+    let validator = MockValidator {};
 
     OverlayProtocol::new(
         overlay_config,
@@ -56,7 +56,7 @@ async fn init_overlay(
 
 async fn spawn_overlay(
     discovery: Arc<Discovery>,
-    overlay: Arc<OverlayProtocol<IdentityContentKey, XorMetric, IdentityValidator>>,
+    overlay: Arc<OverlayProtocol<IdentityContentKey, XorMetric, MockValidator>>,
 ) {
     let (overlay_tx, mut overlay_rx) = mpsc::unbounded_channel();
     let mut discovery_rx = discovery

trin-history/src/validation.rs

@@ -26,14 +26,13 @@ impl Validator<HistoryContentKey> for ChainHistoryValidator {
         match content_key {
             HistoryContentKey::BlockHeader(key) => {
                 let rlp = Rlp::new(content);
-                let header = Header::decode_rlp(&rlp).expect("invalid header");
+                let header = Header::decode_rlp(&rlp)?;
                 let number = format!("0x{:02X}", header.number);
                 let expected_hash = &self
                     .header_oracle
                     .write()
                     .unwrap()
-                    .get_hash_at_height(number)
-                    .unwrap();
+                    .get_hash_at_height(number)?;
                 let actual_hash = &hex::encode(key.block_hash);
                 if actual_hash == expected_hash {
                     Ok(())