feat(validator_node): new state store implementation using RocksDB

Cargo.toml

@@ -28,10 +28,12 @@ members = [
     "dan_layer/indexer_lib",
     "dan_layer/p2p",
     "dan_layer/rpc_state_sync",
+    "dan_layer/state_store_rocksdb",
     "dan_layer/state_store_sqlite",
     "dan_layer/state_tree",
     "dan_layer/storage_sqlite",
     "dan_layer/storage",
+    "dan_layer/storage_tests",
     "dan_layer/tari_bor",
     "dan_layer/template_abi",
     "dan_layer/template_builtin",
@@ -94,6 +96,7 @@ tari_networking = { path = "networking/core" }
 tari_rpc_framework = { path = "networking/rpc_framework" }
 tari_rpc_macros = { path = "networking/rpc_macros" }
 tari_state_store_sqlite = { path = "dan_layer/state_store_sqlite" }
+tari_state_store_rocksdb = { path = "dan_layer/state_store_rocksdb" }
 tari_state_tree = { path = "dan_layer/state_tree" }
 tari_swarm = { path = "networking/swarm" }
 tari_template_abi = { version = "0.7", path = "dan_layer/template_abi" }
@@ -214,6 +217,9 @@ quote = "1.0.7"
 rand = "0.8.5"
 rayon = "1.7.0"
 reqwest = "0.11.16"
+# TODO: use the next release (the one after 0.22.0) when available
+# https://github.com/rust-rocksdb/rust-rocksdb/issues/881
+rocksdb = { git = "https://github.com/rust-rocksdb/rust-rocksdb", rev = "1cf906dc4087f06631820f13855e6b27bd21b972" }
 semver = "1.0"
 serde = { version = "1.0", default-features = false }
 serde_json = "1.0"

applications/tari_dan_app_utilities/config_presets/c_validator_node.toml

@@ -28,6 +28,9 @@
 # The relative path to store persistent data (default = "data/validator_node")
 #data_dir = "data/validator_node"
 
+# The type of database ("rocksdb" or "sqlite") to use (default = "rocksdb")
+#database_type = "rocksdb"
+
 # JSON-RPC listener address (default = "127.0.0.1:18200")
 #json_rpc_address = "127.0.0.1:18200"
 

applications/tari_dan_app_utilities/src/base_layer_scanner.rs

@@ -74,13 +74,13 @@ use crate::template_manager::interface::{TemplateExecutable, TemplateManagerErro
 
 const LOG_TARGET: &str = "tari::dan::base_layer_scanner";
 
-pub fn spawn<TAddr: NodeAddressable + 'static>(
+pub fn spawn<TAddr: NodeAddressable + 'static, TStore: StateStore + Send + 'static>(
     global_db: GlobalDb<SqliteGlobalDbAdapter<TAddr>>,
     base_node_client: GrpcBaseNodeClient,
     epoch_manager: EpochManagerHandle<TAddr>,
     shutdown: ShutdownSignal,
     consensus_constants: ConsensusConstants,
-    shard_store: SqliteStateStore<TAddr>,
+    shard_store: TStore,
     scan_base_layer: bool,
     base_layer_scanning_interval: Duration,
     validator_node_sidechain_id: Option<RistrettoPublicKey>,
@@ -110,7 +110,7 @@ pub fn spawn<TAddr: NodeAddressable + 'static>(
     })
 }
 
-pub struct BaseLayerScanner<TAddr> {
+pub struct BaseLayerScanner<TAddr, TStore> {
     global_db: GlobalDb<SqliteGlobalDbAdapter<TAddr>>,
     last_scanned_height: u64,
     last_scanned_tip: Option<FixedHash>,
@@ -121,7 +121,7 @@ pub struct BaseLayerScanner<TAddr> {
     epoch_manager: EpochManagerHandle<TAddr>,
     shutdown: ShutdownSignal,
     consensus_constants: ConsensusConstants,
-    state_store: SqliteStateStore<TAddr>,
+    state_store: TStore,
     scan_base_layer: bool,
     base_layer_scanning_interval: Duration,
     has_attempted_scan: bool,
@@ -132,14 +132,14 @@ pub struct BaseLayerScanner<TAddr> {
     template_sidechain_id: Option<PublicKey>,
 }
 
-impl<TAddr: NodeAddressable + 'static> BaseLayerScanner<TAddr> {
+impl<TAddr: NodeAddressable + 'static, TStore: StateStore> BaseLayerScanner<TAddr, TStore> {
     pub fn new(
         global_db: GlobalDb<SqliteGlobalDbAdapter<TAddr>>,
         base_node_client: GrpcBaseNodeClient,
         epoch_manager: EpochManagerHandle<TAddr>,
         shutdown: ShutdownSignal,
         consensus_constants: ConsensusConstants,
-        state_store: SqliteStateStore<TAddr>,
+        state_store: TStore,
         scan_base_layer: bool,
         base_layer_scanning_interval: Duration,
         validator_node_sidechain_id: Option<PublicKey>,

applications/tari_indexer/src/bootstrap.rs

@@ -154,7 +154,7 @@ pub async fn spawn_services(
         shutdown.clone(),
         consensus_constants,
         // TODO: Remove coupling between scanner and shard store
-        SqliteStateStore::connect(&format!(
+        SqliteStateStore::<PeerAddress>::connect(&format!(
             "sqlite://{}",
             config.indexer.data_dir.join("unused-shard-store.sqlite").display()
         ))?,

applications/tari_validator_node/Cargo.toml

@@ -39,6 +39,8 @@ tari_rpc_state_sync = { workspace = true }
 tari_bor = { workspace = true, default-features = true }
 tari_consensus = { workspace = true }
 tari_state_store_sqlite = { workspace = true }
+tari_state_store_rocksdb = { workspace = true }
+tari_state_tree = { workspace = true }
 tari_networking = { workspace = true }
 tari_rpc_framework = { workspace = true }
 tari_template_builtin = { workspace = true }