starknet_patricia_storage: spawn blocking tasks for rocksDB reads

Author: nimrod-starkware

crates/starknet_patricia_storage/Cargo.toml

@@ -31,6 +31,7 @@ serde_json.workspace = true
 starknet-types-core.workspace = true
 starknet_api.workspace = true
 thiserror.workspace = true
+tokio.workspace = true
 validator = { workspace = true, features = ["derive"] }
 
 [dev-dependencies]

crates/starknet_patricia_storage/src/errors.rs

@@ -45,5 +45,5 @@ pub enum DeserializationError {
     // TODO(Ariel): This is only used for EdgeNode construction failures (path length etc.), add
     // error types here and use them instead of the general ValueError.
     #[error("Invalid value for deserialization: {0}.")]
-    ValueError(Box<dyn std::error::Error>),
+    ValueError(Box<dyn std::error::Error + Send>),
 }

crates/starknet_patricia_storage/src/rocksdb_storage.rs

@@ -13,6 +13,7 @@ use rust_rocksdb::{
     WriteOptions,
     DB,
 };
+use tokio::task::spawn_blocking;
 
 use crate::storage_trait::{
     AsyncStorage,
@@ -142,22 +143,28 @@ impl Storage for RocksDbStorage {
     type Config = EmptyStorageConfig;
 
     async fn get(&self, key: &DbKey) -> PatriciaStorageResult<Option<DbValue>> {
-        Ok(self.db.get(&key.0)?.map(DbValue))
+        // TODO(Nimrod): Config should indicate whether to spawn a task or not.
+        let db = self.db.clone();
+        let key = key.clone();
+        Ok(spawn_blocking(move || db.get(&key.0).map(|opt| opt.map(DbValue))).await??)
     }
 
     async fn set(&mut self, key: DbKey, value: DbValue) -> PatriciaStorageResult<()> {
         Ok(self.db.put_opt(&key.0, &value.0, &self.options.write_options)?)
     }
 
     async fn mget(&self, keys: &[&DbKey]) -> PatriciaStorageResult<Vec<Option<DbValue>>> {
-        let raw_keys = keys.iter().map(|k| &k.0);
-        let res = self
-            .db
-            .multi_get(raw_keys)
-            .into_iter()
-            .map(|r| r.map(|opt| opt.map(DbValue)))
-            .collect::<Result<_, _>>()?;
-        Ok(res)
+        // TODO(Nimrod): Config should indicate whether to spawn a task or not.
+        let db = self.db.clone();
+        let keys: Vec<Vec<u8>> = keys.iter().map(|k| k.0.clone()).collect();
+        spawn_blocking(move || {
+            let raw_keys = keys.iter().map(|k| k.as_slice());
+            db.multi_get(raw_keys)
+                .into_iter()
+                .map(|r| r.map(|opt| opt.map(DbValue)).map_err(|e| e.into()))
+                .collect::<Result<Vec<_>, PatriciaStorageError>>()
+        })
+        .await?
     }
 
     async fn mset(&mut self, key_to_value: DbHashMap) -> PatriciaStorageResult<()> {

crates/starknet_patricia_storage/src/storage_trait.rs

@@ -7,6 +7,7 @@ use apollo_config::dumping::{ser_param, SerializeConfig};
 use apollo_config::{ParamPath, ParamPrivacyInput, SerializedParam};
 use serde::{Deserialize, Serialize, Serializer};
 use starknet_types_core::felt::Felt;
+use tokio::task::JoinError;
 use validator::Validate;
 
 use crate::errors::DeserializationError;
@@ -37,6 +38,8 @@ pub enum PatriciaStorageError {
     AerospikeStorage(#[from] crate::aerospike_storage::AerospikeStorageError),
     #[error(transparent)]
     Deserialization(#[from] DeserializationError),
+    #[error(transparent)]
+    Join(#[from] JoinError),
     #[cfg(feature = "mdbx_storage")]
     #[error(transparent)]
     Mdbx(#[from] libmdbx::Error),