Add content validation for accepted content

Author: njgheorghita

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};
@@ -116,12 +117,13 @@ pub struct OverlayProtocol<TContentKey, TMetric, TValidator> {
     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>,
+    /// Non-blocking validator that makes requests to this/other overlay networks (or infura) to
+    /// validate accepted content.
+    validator: Arc<Mutex<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 +137,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(),
@@ -156,7 +158,7 @@ where
             protocol.clone(),
             utp_listener_tx.clone(),
             config.enable_metrics,
-            validator,
+            Arc::clone(&validator),
             config.query_timeout,
             config.query_peer_timeout,
             config.query_parallelism,
@@ -176,7 +178,7 @@ where
             utp_listener_tx,
             phantom_content_key: PhantomData,
             phantom_metric: PhantomData,
-            phantom_validator: PhantomData,
+            validator,
         }
     }
 
@@ -253,35 +255,60 @@ where
             ));
         }
 
-        // TODO: Verify overlay content data with an Oracle
-
-        // Temporarily store content key/value pairs to propagate here
-        let mut content_keys_values: Vec<(TContentKey, ByteList)> = Vec::new();
-
-        // Try to store the content into the database and propagate gossip the content
+        // Validate, and try to store the content into the database and propagate gossip the content
         for (content_key, content_value) in content_keys.into_iter().zip(content_values.to_vec()) {
             match TContentKey::try_from(content_key) {
                 Ok(key) => {
-                    // Store accepted content in DB
-                    self.store_overlay_content(&key, content_value.clone());
-                    content_keys_values.push((key, content_value))
+                    // Spawn a task that...
+                    // - Validates accepted content (this step requires a dedicated task since it
+                    // might require non-blocking requests to this/other overlay networks)
+                    // - Checks if content should be stored, and stores it if true
+                    // - Propagate all validated content
+                    let validator = Arc::clone(&self.validator);
+                    let storage = Arc::clone(&self.storage);
+                    let kbuckets = Arc::clone(&self.kbuckets);
+                    let request_tx = self.request_tx.clone();
+
+                    tokio::spawn(async move {
+                        // Validated received content
+                        let mut lock = validator.lock().await;
+                        if let Err(err) = lock.validate_content(&key, &content_value.to_vec()).await
+                        {
+                            // Skip storing & propagating content if it's not valid
+                            error!("Unable to validate received content: {err:?}");
+                            return;
+                        };
+
+                        // Check if data should be stored, and store if true.
+                        let _ = storage
+                            .write()
+                            .store_if_should(&key, &content_value.to_vec());
+
+                        // Propagate all validated content, whether or not it was stored.
+                        OverlayProtocol::<TContentKey, TMetric, TValidator>::propagate_gossip(
+                            (key, content_value),
+                            kbuckets,
+                            request_tx,
+                        );
+                    });
                 }
                 Err(err) => {
-                    return Err(anyhow!(
-                        "Unexpected error while decoding overlay content key: {err}"
-                    ));
+                    warn!("Unexpected error while decoding overlay content key: {err}");
+                    continue;
                 }
             }
         }
-        // Propagate gossip accepted content
-        self.propagate_gossip(content_keys_values)?;
         Ok(())
     }
 
     /// Propagate gossip accepted content via OFFER/ACCEPT:
-    fn propagate_gossip(&self, content: Vec<(TContentKey, ByteList)>) -> anyhow::Result<()> {
+    fn propagate_gossip(
+        content_key_value: (TContentKey, ByteList),
+        kbuckets: Arc<RwLock<KBucketsTable<NodeId, Node>>>,
+        request_tx: UnboundedSender<OverlayRequest>,
+    ) {
         // Get all nodes from overlay routing table
-        let kbuckets = self.kbuckets.read();
+        let kbuckets = kbuckets.read();
         let all_nodes: Vec<&kbucket::Node<NodeId, Node>> = kbuckets
             .buckets_iter()
             .map(|kbucket| {
@@ -297,37 +324,44 @@ where
         let mut enrs_and_content: HashMap<String, Vec<RawContentKey>> = HashMap::new();
 
         // Filter all nodes from overlay routing table where XOR_distance(content_id, nodeId) < node radius
-        for content_key_value in content {
-            let interested_enrs: Vec<Enr> = all_nodes
-                .clone()
-                .into_iter()
-                .filter(|node| {
-                    XorMetric::distance(
-                        &content_key_value.0.content_id(),
-                        &node.key.preimage().raw(),
-                    ) < node.value.data_radius()
-                })
-                .map(|node| node.value.enr())
-                .collect();
-
-            // Continue if no nodes are interested in the content
-            if interested_enrs.is_empty() {
-                debug!("No nodes interested in neighborhood gossip: content_key={} num_nodes_checked={}",
-                    hex_encode(content_key_value.0.into()), all_nodes.len());
-                continue;
-            }
+        let interested_enrs: Vec<Enr> = all_nodes
+            .clone()
+            .into_iter()
+            .filter(|node| {
+                XorMetric::distance(
+                    &content_key_value.0.content_id(),
+                    &node.key.preimage().raw(),
+                ) < node.value.data_radius()
+            })
+            .map(|node| node.value.enr())
+            .collect();
 
-            // Get log2 random ENRs to gossip
-            let random_enrs = log2_random_enrs(interested_enrs)?;
+        // Continue if no nodes are interested in the content
+        if interested_enrs.is_empty() {
+            debug!(
+                "No nodes interested in neighborhood gossip: content_key={} num_nodes_checked={}",
+                hex_encode(content_key_value.0.into()),
+                all_nodes.len()
+            );
+            return;
+        }
 
-            // Temporarily store all randomly selected nodes with the content of interest.
-            // We want this so we can offer all the content to interested node in one request.
-            for enr in random_enrs {
-                enrs_and_content
-                    .entry(enr.to_base64())
-                    .or_default()
-                    .push(content_key_value.clone().0.into());
+        // Get log2 random ENRs to gossip
+        let random_enrs = match log2_random_enrs(interested_enrs) {
+            Ok(val) => val,
+            Err(msg) => {
+                debug!("No available enrs for gossip: {msg}");
+                return;
             }
+        };
+
+        // Temporarily store all randomly selected nodes with the content of interest.
+        // We want this so we can offer all the content to interested node in one request.
+        for enr in random_enrs {
+            enrs_and_content
+                .entry(enr.to_base64())
+                .or_default()
+                .push(content_key_value.clone().0.into());
         }
 
         // Create and send OFFER overlay request to the interested nodes
@@ -351,24 +385,10 @@ where
                 None,
             );
 
-            if let Err(err) = self.request_tx.send(overlay_request) {
+            if let Err(err) = request_tx.send(overlay_request) {
                 error!("Unable to send OFFER request to overlay: {err}.")
             }
         }
-        Ok(())
-    }
-
-    /// Try to store overlay content into database
-    fn store_overlay_content(&self, content_key: &TContentKey, value: ByteList) {
-        let should_store = self.storage.read().should_store(content_key);
-        match should_store {
-            Ok(_) => {
-                if let Err(err) = self.storage.write().store(content_key, &value.into()) {
-                    warn!("Unable to store accepted content: {err}");
-                }
-            }
-            Err(err) => error!("Unable to determine whether to store accepted content: {err}"),
-        }
     }
 
     /// Returns a vector of all ENR node IDs of nodes currently contained in the routing table.

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,
@@ -1317,6 +1318,7 @@ where
             Content::Content(content) => {
                 self.process_received_content(content.clone(), request)
                     .await;
+                // huh?
                 // 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());
@@ -1344,19 +1346,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 +1999,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/src/portalnet/storage.rs

@@ -136,6 +136,21 @@ impl PortalStorage {
         }
     }
 
+    /// Public method for automatically storing content after a `should_store` check.
+    pub fn store_if_should(
+        &mut self,
+        key: &impl OverlayContentKey,
+        value: &Vec<u8>,
+    ) -> Result<bool, PortalStorageError> {
+        match self.should_store(key)? {
+            true => {
+                let _ = self.store(key, value)?;
+                Ok(true)
+            }
+            false => Ok(false),
+        }
+    }
+
     /// Public method for storing a given value for a given content-key.
     pub fn store(
         &mut self,

trin-core/tests/overlay.rs

@@ -21,6 +21,7 @@ use discv5::Discv5Event;
 use ethereum_types::U256;
 use parking_lot::RwLock;
 use tokio::{
+    sync::Mutex,
     sync::{mpsc, mpsc::unbounded_channel},
     time::{self, Duration},
 };
@@ -40,7 +41,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/events.rs

@@ -15,6 +15,7 @@ impl StateEvents {
     pub async fn start(mut self) {
         loop {
             tokio::select! {
+                // blocking async!
                 Some(talk_request) = self.event_rx.recv() => {
                     self.handle_state_talk_request(talk_request).await;
                 }

trin-state/src/network.rs

@@ -5,6 +5,8 @@ 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 +44,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,