Improvement: chainsync asynchronism

crates/discovery/Cargo.toml

@@ -9,6 +9,7 @@ alloy = { workspace = true }
 anyhow = { workspace = true }
 clap = { workspace = true }
 env_logger = { workspace = true }
+futures = { workspace = true }
 log = { workspace = true }
 redis = { workspace = true }
 redis-test = { workspace = true }

crates/discovery/src/chainsync/sync.rs

@@ -1,6 +1,7 @@
 use crate::store::node_store::NodeStore;
 use alloy::primitives::Address;
 use anyhow::Error;
+use futures::StreamExt;
 use log::error;
 use shared::models::node::DiscoveryNode;
 use shared::web3::contracts::core::builder::Contracts;
@@ -37,10 +38,8 @@ impl ChainSync {
     async fn sync_single_node(
         node_store: Arc<NodeStore>,
         contracts: Arc<Contracts>,
-        node: DiscoveryNode,
+        mut node: DiscoveryNode,
     ) -> Result<(), Error> {
-        let mut n = node.clone();
-
         // Safely parse provider_address and node_address
         let provider_address = Address::from_str(&node.provider_address).map_err(|e| {
             eprintln!("Failed to parse provider address: {}", e);
@@ -52,40 +51,37 @@ impl ChainSync {
             anyhow::anyhow!("Invalid node address")
         })?;
 
-        let node_info = contracts
-            .compute_registry
-            .get_node(provider_address, node_address)
-            .await
-            .map_err(|e| {
-                eprintln!("Error retrieving node info: {}", e);
-                anyhow::anyhow!("Failed to retrieve node info")
-            })?;
+        let (node_info_result, provider_info_result, is_blacklisted_result) = tokio::join!(
+            contracts
+                .compute_registry
+                .get_node(provider_address, node_address),
+            contracts.compute_registry.get_provider(provider_address),
+            contracts
+                .compute_pool
+                .is_node_blacklisted(node.node.compute_pool_id, node_address),
+        );
 
-        let provider_info = contracts
-            .compute_registry
-            .get_provider(provider_address)
-            .await
-            .map_err(|e| {
-                eprintln!("Error retrieving provider info: {}", e);
-                anyhow::anyhow!("Failed to retrieve provider info")
-            })?;
+        let node_info = node_info_result.map_err(|e| {
+            eprintln!("Error retrieving node info: {}", e);
+            anyhow::anyhow!("Failed to retrieve node info")
+        })?;
 
-        let (is_active, is_validated) = node_info;
-        n.is_active = is_active;
-        n.is_validated = is_validated;
-        n.is_provider_whitelisted = provider_info.is_whitelisted;
+        let provider_info = provider_info_result.map_err(|e| {
+            eprintln!("Error retrieving provider info: {}", e);
+            anyhow::anyhow!("Failed to retrieve provider info")
+        })?;
 
-        // Handle potential errors from async calls
-        let is_blacklisted = contracts
-            .compute_pool
-            .is_node_blacklisted(node.node.compute_pool_id, node_address)
-            .await
-            .map_err(|e| {
-                eprintln!("Error checking if node is blacklisted: {}", e);
-                anyhow::anyhow!("Failed to check blacklist status")
-            })?;
-        n.is_blacklisted = is_blacklisted;
-        match node_store.update_node(n) {
+        let is_blacklisted = is_blacklisted_result.map_err(|e| {
+            eprintln!("Error checking if node is blacklisted: {}", e);
+            anyhow::anyhow!("Failed to check blacklist status")
+        })?;
+
+        let (is_active, is_validated) = node_info;
+        node.is_active = is_active;
+        node.is_validated = is_validated;
+        node.is_provider_whitelisted = provider_info.is_whitelisted;
+        node.is_blacklisted = is_blacklisted;
+        match node_store.update_node(node) {
             Ok(_) => (),
             Err(e) => {
                 error!("Error updating node: {}", e);
@@ -110,11 +106,17 @@ impl ChainSync {
                         let nodes = node_store_clone.get_nodes();
                         match nodes {
                             Ok(nodes) => {
-                                for node in nodes {
-                                    if let Err(e) = ChainSync::sync_single_node(node_store_clone.clone(), contracts_clone.clone(), node).await {
-                                        error!("Error syncing node: {}", e);
-                                    }
-                                }
+                                futures::stream::iter(nodes)
+                                    .for_each_concurrent(10, |node| {
+                                        let node_store = node_store_clone.clone();
+                                        let contracts = contracts_clone.clone();
+                                        async move {
+                                            if let Err(e) = ChainSync::sync_single_node(node_store, contracts, node).await {
+                                                error!("Error syncing node: {}", e);
+                                            }
+                                        }
+                                    })
+                                    .await;
                                 // Update the last chain sync time
                                 let mut last_sync = last_chain_sync.lock().await;
                                 *last_sync = Some(SystemTime::now());

crates/shared/src/web3/contracts/implementations/compute_pool_contract.rs

@@ -173,7 +173,7 @@ impl ComputePool {
         &self,
         pool_id: u32,
         node: Address,
-    ) -> Result<bool, Box<dyn std::error::Error>> {
+    ) -> Result<bool, Box<dyn std::error::Error + Send + Sync>> {
         let arg_pool_id: U256 = U256::from(pool_id);
         let result = self
             .instance
@@ -184,7 +184,13 @@ impl ComputePool {
             )?
             .call()
             .await?;
-        Ok(result.first().unwrap().as_bool().unwrap())
+        let first_value = result
+            .first()
+            .ok_or("Empty response when checking if node is blacklisted")?;
+        let is_blacklisted = first_value
+            .as_bool()
+            .ok_or("Expected a boolean value in response for node blacklisted check")?;
+        Ok(is_blacklisted)
     }
 
     pub async fn get_blacklisted_nodes(

crates/shared/src/web3/contracts/implementations/compute_registry_contract.rs

@@ -22,7 +22,7 @@ impl ComputeRegistryContract {
     pub async fn get_provider(
         &self,
         address: Address,
-    ) -> Result<ComputeProvider, Box<dyn std::error::Error>> {
+    ) -> Result<ComputeProvider, Box<dyn std::error::Error + Send + Sync>> {
         let provider_response = self
             .instance
             .instance()