Merge 07704fedc9 into baf0521c99

node/storage-async/src/lib.rs

@@ -1,7 +1,7 @@
 #[macro_use]
 extern crate tracing;
 
-use anyhow::bail;
+use anyhow::{anyhow, bail};
 use shared_types::{
     Chunk, ChunkArray, ChunkArrayWithProof, DataRoot, FlowProof, FlowRangeProof, Transaction,
 };
@@ -136,17 +136,21 @@ impl Store {
         let store = self.store.clone();
         let (tx, rx) = oneshot::channel();
 
-        self.executor.spawn_blocking(
+        self.executor
-            move || {
+            .spawn_blocking_handle(
-                // FIXME(zz): Not all functions need `write`. Refactor store usage.
+                move || {
-                let res = f(&*store);
+                    // FIXME(zz): Not all functions need `write`. Refactor store usage.
+                    let res = f(&*store);
 
-                if tx.send(res).is_err() {
+                    if tx.send(res).is_err() {
-                    error!("Unable to complete async storage operation: the receiver dropped");
+                        error!("Unable to complete async storage operation: the receiver dropped");
-                }
+                    }
-            },
+                },
-            WORKER_TASK_NAME,
+                WORKER_TASK_NAME,
-        );
+            )
+            .ok_or(anyhow!("Unable to spawn async storage work"))?
+            .await
+            .map_err(|e| anyhow!("join error: e={:?}", e))?;
 
         rx.await
             .unwrap_or_else(|_| bail!(error::Error::Custom("Receiver error".to_string())))

tests/config/node_config.py

@@ -1,3 +1,5 @@
+import os
+
 from web3 import Web3
 
 ZGS_CONFIG = {
@@ -22,6 +24,8 @@ ZGS_CONFIG = {
     }
 }
 
+CONFIG_DIR = os.path.dirname(__file__)
+ZGS_KEY_FILE = os.path.join(CONFIG_DIR, "zgs", "network", "key")
 ZGS_NODEID = "16Uiu2HAmLkGFUbNFYdhuSbTQ5hmnPjFXx2zUDtwQ2uihHpN9YNNe"
 
 BSC_CONFIG = dict(

tests/config/zgs/network/enr.dat

@@ -1 +0,0 @@
-enr:-Ly4QJZwz9htAorBIx_otqoaRFPohX7NQJ31iBB6mcEhBiuPWsOnigc1ABQsg6tLU1OirQdLR6aEvv8SlkkfIbV72T8CgmlkgnY0gmlwhH8AAAGQbmV0d29ya19pZGVudGl0eZ8oIwAAAAAAADPyz8cpvYcPpUtQMmYOBrTPKn-UAAIAiXNlY3AyNTZrMaEDeDdgnDgLPkxNxB39jKb9f1Na30t6R9vVolpTk5zu-hODdGNwgir4g3VkcIIq-A

tests/network_discovery_test.py

@@ -3,7 +3,7 @@
 import os
 import time
 
-from config.node_config import ZGS_NODEID
+from config.node_config import ZGS_KEY_FILE, ZGS_NODEID
 from test_framework.test_framework import TestFramework
 from utility.utils import p2p_port
 
@@ -17,13 +17,9 @@ class NetworkDiscoveryTest(TestFramework):
         self.num_nodes = 3
 
         # setup for node 0 as bootnode
-        tests_dir = os.path.dirname(__file__)
+        self.zgs_node_key_files = [ZGS_KEY_FILE]
-        network_dir = os.path.join(tests_dir, "config", "zgs", "network")
         bootnode_port = p2p_port(0)
         self.zgs_node_configs[0] = {
-            # load pre-defined keypair
-            "network_dir": network_dir,
-
             # enable UDP discovery relevant configs
             "network_enr_address": "127.0.0.1",
             "network_enr_tcp_port": bootnode_port,

tests/network_discovery_upgrade_test.py

@@ -3,7 +3,7 @@
 import os
 import time
 
-from config.node_config import ZGS_NODEID
+from config.node_config import ZGS_KEY_FILE, ZGS_NODEID
 from test_framework.test_framework import TestFramework
 from utility.utils import p2p_port
 
@@ -17,13 +17,9 @@ class NetworkDiscoveryUpgradeTest(TestFramework):
         self.num_nodes = 2
 
         # setup for node 0 as bootnode
-        tests_dir = os.path.dirname(__file__)
+        self.zgs_node_key_files = [ZGS_KEY_FILE]
-        network_dir = os.path.join(tests_dir, "config", "zgs", "network")
         bootnode_port = p2p_port(0)
         self.zgs_node_configs[0] = {
-            # load pre-defined keypair
-            "network_dir": network_dir,
-
             # enable UDP discovery relevant configs
             "network_enr_address": "127.0.0.1",
             "network_enr_tcp_port": bootnode_port,

tests/test_framework/test_framework.py

@@ -55,6 +55,7 @@ class TestFramework:
         self.lifetime_seconds = 3600
         self.launch_wait_seconds = 1
         self.num_deployed_contracts = 0
+        self.zgs_node_key_files = []
 
         # Set default binary path
         binary_ext = ".exe" if is_windows_platform() else ""
@@ -190,6 +191,10 @@ class TestFramework:
             else:
                 updated_config = {}
 
+            zgs_node_key_file = None
+            if i < len(self.zgs_node_key_files):
+                zgs_node_key_file = self.zgs_node_key_files[i]
+
             assert os.path.exists(self.zgs_binary), (
                 "%s should be exist" % self.zgs_binary
             )
@@ -202,6 +207,7 @@ class TestFramework:
                 self.mine_contract.address(),
                 self.reward_contract.address(),
                 self.log,
+                key_file=zgs_node_key_file,
             )
             self.nodes.append(node)
             node.setup_config()

tests/test_framework/zgs_node.py

@@ -24,6 +24,7 @@ class ZgsNode(TestNode):
         log,
         rpc_timeout=10,
         libp2p_nodes=None,
+        key_file=None,
     ):
         local_conf = ZGS_CONFIG.copy()
         if libp2p_nodes is None:
@@ -54,6 +55,7 @@ class ZgsNode(TestNode):
         # Overwrite with personalized configs.
         update_config(local_conf, updated_config)
         data_dir = os.path.join(root_dir, "zgs_node" + str(index))
+        self.key_file = key_file
         rpc_url = "http://" + rpc_listen_address
         super().__init__(
             NodeType.Zgs,
@@ -68,10 +70,16 @@ class ZgsNode(TestNode):
 
     def setup_config(self):
         os.mkdir(self.data_dir)
+
         log_config_path = os.path.join(self.data_dir, self.config["log_config_file"])
         with open(log_config_path, "w") as f:
             f.write("trace,hyper=info,h2=info")
 
+        if self.key_file is not None:
+            network_dir = os.path.join(self.data_dir, "network")
+            os.mkdir(network_dir)
+            shutil.copy(self.key_file, network_dir)
+
         initialize_toml_config(self.config_file, self.config)
 
     def wait_for_rpc_connection(self):