Spawn non-blocking task for content validation

Author: njgheorghita

newsfragments/373.fixed.md

@@ -0,0 +1 @@
+Remove blocking async from overlay service loop via spawning new tasks for utp & content validation.

trin-core/src/portalnet/overlay.rs

@@ -50,6 +50,7 @@ use rand::seq::IteratorRandom;
 use ssz::{Decode, Encode};
 use ssz_types::VariableList;
 use tokio::sync::mpsc::UnboundedSender;
+use tokio::sync::Mutex;
 use tracing::{debug, warn};
 
 pub use super::overlay_service::{OverlayRequestError, RequestDirection};
@@ -121,7 +122,7 @@ pub struct OverlayProtocol<TContentKey, TMetric, TValidator> {
 }
 
 impl<
-        TContentKey: OverlayContentKey + Send + Sync,
+        TContentKey: 'static + OverlayContentKey + Send + Sync,
         TMetric: Metric + Send + Sync,
         TValidator: 'static + Validator<TContentKey> + Send + Sync,
     > OverlayProtocol<TContentKey, TMetric, TValidator>
@@ -135,7 +136,7 @@ where
         storage: Arc<RwLock<PortalStorage>>,
         data_radius: U256,
         protocol: ProtocolId,
-        validator: TValidator,
+        validator: Arc<Mutex<TValidator>>,
     ) -> Self {
         let kbuckets = Arc::new(RwLock::new(KBucketsTable::new(
             discovery.local_enr().node_id().into(),

trin-core/src/portalnet/overlay_service.rs

@@ -68,6 +68,7 @@ use ssz::Encode;
 use ssz_types::{BitList, VariableList};
 use thiserror::Error;
 use tokio::sync::mpsc::{self, UnboundedReceiver, UnboundedSender};
+use tokio::sync::Mutex;
 
 /// Maximum number of ENRs in response to FindNodes.
 pub const FIND_NODES_MAX_NODES: usize = 32;
@@ -317,11 +318,11 @@ pub struct OverlayService<TContentKey, TMetric, TValidator> {
     /// Metrics reporting component
     metrics: Option<OverlayMetrics>,
     /// Validator for overlay network content.
-    validator: TValidator,
+    validator: Arc<Mutex<TValidator>>,
 }
 
 impl<
-        TContentKey: OverlayContentKey + Send + Sync,
+        TContentKey: 'static + OverlayContentKey + Send + Sync,
         TMetric: Metric + Send + Sync,
         TValidator: 'static + Validator<TContentKey> + Send + Sync,
     > OverlayService<TContentKey, TMetric, TValidator>
@@ -343,7 +344,7 @@ where
         protocol: ProtocolId,
         utp_listener_sender: UnboundedSender<UtpListenerRequest>,
         enable_metrics: bool,
-        validator: TValidator,
+        validator: Arc<Mutex<TValidator>>,
         query_timeout: Duration,
         query_peer_timeout: Duration,
         query_parallelism: usize,
@@ -486,7 +487,7 @@ where
 
                         // Perform background processing.
                         match response.response {
-                            Ok(response) => self.process_response(response, active_request.destination, active_request.request, active_request.query_id).await,
+                            Ok(response) => self.process_response(response, active_request.destination, active_request.request, active_request.query_id),
                             Err(error) => self.process_request_failure(response.request_id, active_request.destination, error),
                         }
 
@@ -1108,7 +1109,7 @@ where
     }
 
     /// Processes a response to an outgoing request from some source node.
-    async fn process_response(
+    fn process_response(
         &mut self,
         response: Response,
         source: Enr,
@@ -1171,7 +1172,6 @@ where
                     }
                 };
                 self.process_content(content, source, find_content_request, query_id)
-                    .await
             }
             Response::Accept(accept) => {
                 let offer_request = match request {
@@ -1182,18 +1182,15 @@ where
                     }
                 };
 
-                if let Err(err) = self
-                    .process_accept(accept, source, offer_request.content_keys)
-                    .await
-                {
+                if let Err(err) = self.process_accept(accept, source, offer_request.content_keys) {
                     error!("Error processing ACCEPT response in overlay service: {err}")
                 }
             }
         }
     }
 
     /// Process ACCEPT response
-    pub async fn process_accept(
+    pub fn process_accept(
         &self,
         response: Accept,
         enr: Enr,
@@ -1227,17 +1224,17 @@ where
         self.utp_listener_tx
             .send(utp_request).map_err(|err| anyhow!("Unable to send Connect request to UtpListener when processing ACCEPT message: {err}"))?;
 
-        let mut conn = rx.await?;
-        // Handle STATE packet for SYN
-        let mut buf = [0; BUF_SIZE];
-        conn.recv(&mut buf).await?;
-
         let content_items = self.provide_requested_content(&response, content_keys_offered)?;
 
         let content_payload = ContentPayloadList::new(content_items)
             .map_err(|err| anyhow!("Unable to build content payload: {err:?}"))?;
 
         tokio::spawn(async move {
+            let mut conn = rx.await.unwrap();
+            // Handle STATE packet for SYN
+            let mut buf = [0; BUF_SIZE];
+            conn.recv(&mut buf).await.unwrap();
+
             // send the content to the acceptor over a uTP stream
             if let Err(err) = conn.send_to(&content_payload.as_ssz_bytes()).await {
                 warn!("Error sending content {err}");
@@ -1297,7 +1294,7 @@ where
     }
 
     /// Processes a Content response.
-    async fn process_content(
+    fn process_content(
         &mut self,
         content: Content,
         source: Enr,
@@ -1315,8 +1312,7 @@ where
                 self.protocol, id
             ),
             Content::Content(content) => {
-                self.process_received_content(content.clone(), request)
-                    .await;
+                self.process_received_content(content.clone(), request);
                 // TODO: Should we only advance the query if the content has been validated?
                 if let Some(query_id) = query_id {
                     self.advance_find_content_query_with_content(&query_id, source, content.into());
@@ -1332,7 +1328,7 @@ where
         }
     }
 
-    async fn process_received_content(&mut self, content: ByteList, request: FindContent) {
+    fn process_received_content(&mut self, content: ByteList, request: FindContent) {
         let content_key = match TContentKey::try_from(request.content_key) {
             Ok(val) => val,
             Err(msg) => {
@@ -1344,19 +1340,23 @@ where
         match should_store {
             Ok(val) => {
                 if val {
-                    // validate content before storing
-                    if let Err(err) = self
-                        .validator
-                        .validate_content(&content_key, &content.to_vec())
-                        .await
-                    {
-                        error!("Unable to validate received content: {err:?}");
-                        return;
-                    };
+                    let validator = Arc::clone(&self.validator);
+                    let storage = Arc::clone(&self.storage);
+                    // Spawn task that validates content before storing.
+                    // Allows for non-blocking requests to this/other overlay services.
+                    tokio::spawn(async move {
+                        let mut lock = validator.lock().await;
+                        if let Err(err) =
+                            lock.validate_content(&content_key, &content.to_vec()).await
+                        {
+                            error!("Unable to validate received content: {err:?}");
+                            return;
+                        };
 
-                    if let Err(err) = self.storage.write().store(&content_key, &content.into()) {
-                        error!("Content received, but not stored: {err}")
-                    }
+                        if let Err(err) = storage.write().store(&content_key, &content.into()) {
+                            error!("Content received, but not stored: {err}")
+                        }
+                    });
                 } else {
                     debug!(
                         "Content received, but not stored: Content is already stored or its distance falls outside current radius."
@@ -1993,7 +1993,7 @@ mod tests {
         let (request_tx, request_rx) = mpsc::unbounded_channel();
         let (response_tx, response_rx) = mpsc::unbounded_channel();
         let metrics = None;
-        let validator = MockValidator {};
+        let validator = Arc::new(Mutex::new(MockValidator {}));
 
         let service = OverlayService {
             discovery,

trin-core/tests/overlay.rs

@@ -21,7 +21,7 @@ use discv5::Discv5Event;
 use ethereum_types::U256;
 use parking_lot::RwLock;
 use tokio::{
-    sync::{mpsc, mpsc::unbounded_channel},
+    sync::{mpsc, mpsc::unbounded_channel, Mutex},
     time::{self, Duration},
 };
 
@@ -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 = MockValidator {};
+    let validator = Arc::new(Mutex::new(MockValidator {}));
 
     OverlayProtocol::new(
         overlay_config,

trin-history/src/network.rs

@@ -3,6 +3,7 @@ use std::sync::{Arc, RwLock as StdRwLock};
 
 use parking_lot::RwLock;
 use tokio::sync::mpsc::UnboundedSender;
+use tokio::sync::Mutex;
 
 use trin_core::{
     portalnet::{
@@ -41,7 +42,7 @@ impl HistoryNetwork {
             ..Default::default()
         };
         let storage = Arc::new(RwLock::new(PortalStorage::new(storage_config).unwrap()));
-        let validator = ChainHistoryValidator { header_oracle };
+        let validator = Arc::new(Mutex::new(ChainHistoryValidator { header_oracle }));
         let overlay = OverlayProtocol::new(
             config,
             discovery,

trin-state/src/network.rs

@@ -5,6 +5,7 @@ use eth_trie::EthTrie;
 use log::{debug, error};
 use parking_lot::RwLock;
 use tokio::sync::mpsc::UnboundedSender;
+use tokio::sync::Mutex;
 use trin_core::{
     portalnet::{
         discovery::Discovery,
@@ -42,9 +43,9 @@ impl StateNetwork {
         let trie = EthTrie::new(Arc::new(triedb));
 
         let storage = Arc::new(RwLock::new(PortalStorage::new(storage_config).unwrap()));
-        let validator = StateValidator {
+        let validator = Arc::new(Mutex::new(StateValidator {
             header_oracle: HeaderOracle::default(),
-        };
+        }));
         let config = OverlayConfig {
             bootnode_enrs: portal_config.bootnode_enrs.clone(),
             enable_metrics: portal_config.enable_metrics,