Merge 07704fedc9 into 3fd800275a

node/network/src/config.rs

@@ -157,8 +157,8 @@ impl Default for Config {
         let filter_rate_limiter = Some(
             discv5::RateLimiterBuilder::new()
                 .total_n_every(300, Duration::from_secs(1)) // Allow bursts, average 300 per second
-                .ip_n_every(300, Duration::from_secs(1)) // Allow bursts, average 300 per second
+                .ip_n_every(9, Duration::from_secs(1)) // Allow bursts, average 9 per second
-                .node_n_every(300, Duration::from_secs(1)) // Allow bursts, average 300 per second
+                .node_n_every(8, Duration::from_secs(1)) // Allow bursts, average 8 per second
                 .build()
                 .expect("The total rate limit has been specified"),
         );

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())))