Single writer/multiple reader MMR store

Author: m-kus

Cargo.toml

@@ -6,9 +6,13 @@ members = ["crates/*"]
 cairo-vm = { git = "https://github.com/maciejka/cairo-vm", rev = "19d8a07ce9799a8af9db6f8a14a8accaad900214" }
 
 [workspace.dependencies]
+# Accumulators
+accumulators = { git = "https://github.com/maciejka/rust-accumulators", rev = "1fbc79a472f1659f0aa0213de95a914086732204", features = ["blake", "memory", "mmr"]}
+
 # Async runtime
 tokio = { version = "1.36", features = ["full"] }
 reqwest = { version = "0.12", features = ["json", "gzip", "brotli", "zstd"] }
+async-trait = "0.1"
 
 # Bitcoin
 bitcoin = { version = "0.32.6", features = ["serde"] }
@@ -19,7 +23,7 @@ jsonrpsee = { version = "0.25.1", features = ["http-client", "async-client"] }
 base64 = "0.21"
 
 # Storage
-libmdbx = "0.3"
+sqlx = { version = "0.8", features = ["runtime-tokio", "sqlite"] }
 
 # CLI
 clap = { version = "4.5", features = ["derive", "env"] }

crates/raito-bridge-node/Cargo.toml

@@ -4,6 +4,8 @@ version = "0.1.0"
 edition = "2021"
 
 [dependencies]
+# Accumulators
+accumulators.workspace = true
 # Core SPV functionality
 raito-spv-mmr = { path = "../raito-spv-mmr", features = ["sqlite"] }
 # Bitcoin client
@@ -12,14 +14,15 @@ raito-spv-client = { path = "../raito-spv-client" }
 raito-spv-verify = { path = "../raito-spv-verify" }
 # Async runtime
 tokio.workspace = true
+async-trait.workspace = true
 # Web framework
 axum = "0.7"
 tower-http = { version = "0.5", features = ["trace", "cors", "compression-gzip"] }
 
 # Bitcoin RPC and types (re-exported from raito-spv-mmr but needed for specific features)
 bitcoin.workspace = true
 # Storage
-libmdbx.workspace = true
+sqlx.workspace = true
 # CLI
 clap.workspace = true
 dotenv.workspace = true

crates/raito-bridge-node/src/app.rs

@@ -1,9 +1,9 @@
 //! Application server and client for managing MMR accumulator operations via async message passing.
 
-use std::path::PathBuf;
+use std::{path::PathBuf, sync::Arc};
 
+use accumulators::hasher::stark_blake::StarkBlakeHasher;
 use bitcoin::{block::Header as BlockHeader, Txid};
-use raito_bitcoin_client::BitcoinClient;
 use tokio::sync::{broadcast, mpsc, oneshot};
 use tracing::{error, info};
 
@@ -14,6 +14,8 @@ use raito_spv_mmr::{
 };
 use raito_spv_verify::TransactionInclusionProof;
 
+use crate::mmr_store::MMRStore;
+
 /// Request sent to the application server via the API channel
 pub struct ApiRequest {
     /// The body of the API request containing the specific operation
@@ -31,8 +33,6 @@ pub enum ApiRequestBody {
     /// Get MMR sparse roots for a given chain height (optional)
     /// The chain height is the number of blocks in the MMR minus one
     GetSparseRoots(Option<u32>),
-    /// Add a new block header to the MMR
-    AddBlock(BlockHeader),
     /// Generate an inclusion proof for a block at the given height and chain height (optional)
     GenerateBlockProof((u32, Option<u32>)),
     /// Get a Bitcoin block header by height
@@ -43,8 +43,6 @@ pub enum ApiRequestBody {
 
 /// Response body for API requests containing the result data
 pub enum ApiResponseBody {
-    /// Empty response
-    Empty,
     /// Response containing the current block count
     GetBlockCount(u32),
     /// Response containing the sparse roots for a given block count
@@ -99,7 +97,12 @@ impl AppServer {
         info!("App server started");
 
         // We need to specify mmr_id to have deterministic keys in the database
-        let mut mmr = BlockMMR::from_file(&self.config.db_path, "blocks").await?;
+        let mmr_id = Some("blocks".to_string());
+        let store = Arc::new(
+            MMRStore::multiple_concurrent_readers(&self.config.db_path, mmr_id.clone()).await?,
+        );
+        let hasher = Arc::new(StarkBlakeHasher::default());
+        let mmr = BlockMMR::new(store, hasher, mmr_id);
 
         loop {
             tokio::select! {
@@ -117,30 +120,9 @@ impl AppServer {
                             let res = mmr.generate_proof(block_height, chain_height).await.map(|proof| ApiResponseBody::GenerateBlockProof(proof));
                             req.tx_response.send(res).map_err(|_| anyhow::anyhow!("Failed to send response to GenerateBlockProof request"))?;
                         }
-                        ApiRequestBody::AddBlock(block_header) => {
-                            // This is a local-only method, so we treat errors differently here
-                            mmr.add_block_header(&block_header).await?;
-                            let res = Ok(ApiResponseBody::Empty);
-                            req.tx_response.send(res).map_err(|_| anyhow::anyhow!("Failed to send response to AddBlock request"))?;
-                        }
                         ApiRequestBody::GetBlockHeader(block_height) => {
-                            let res = async {
-                                let bitcoin_client = BitcoinClient::new(
-                                    self.config.bitcoin_rpc_url.clone(),
-                                    self.config.bitcoin_rpc_userpwd.clone(),
-                                )?;
-
-                                let (block_header, _block_hash) = bitcoin_client
-                                    .get_block_header_by_height(block_height)
-                                    .await?;
-
-                                Ok(ApiResponseBody::GetBlockHeader(block_header))
-                            }
-                            .await;
-
-                            req.tx_response
-                                .send(res)
-                                .map_err(|_| anyhow::anyhow!("Failed to send response to GetBlockHeader request"))?;
+                            let res = mmr.get_block_headers(block_height, 1).await.map(|block_headers| ApiResponseBody::GetBlockHeader(block_headers[0]));
+                            req.tx_response.send(res).map_err(|_| anyhow::anyhow!("Failed to send response to GetBlockHeader request"))?;
                         }
                         ApiRequestBody::GetTransactionProof(txid) => {
                             let res = fetch_transaction_proof(
@@ -227,17 +209,6 @@ impl AppClient {
         .await
     }
 
-    pub async fn add_block(&self, block_header: BlockHeader) -> Result<(), anyhow::Error> {
-        self.send_request(
-            ApiRequestBody::AddBlock(block_header),
-            |response| match response {
-                ApiResponseBody::Empty => Some(()),
-                _ => None,
-            },
-        )
-        .await
-    }
-
     pub async fn generate_block_proof(
         &self,
         block_height: u32,

crates/raito-bridge-node/src/indexer.rs

@@ -1,18 +1,20 @@
 //! Bitcoin blockchain indexer that builds MMR accumulator and generates sparse roots for new blocks.
 
+use std::{path::PathBuf, sync::Arc};
+
+use accumulators::hasher::stark_blake::StarkBlakeHasher;
+use raito_spv_mmr::block_mmr::BlockMMR;
 use tokio::sync::broadcast;
 use tracing::{error, info};
 
 use raito_bitcoin_client::BitcoinClient;
 
-use crate::app::AppClient;
+use crate::mmr_store::MMRStore;
 
 /// Bitcoin block indexer that builds MMR accumulator and generates sparse roots
 pub struct Indexer {
     /// Indexer configuration
     config: IndexerConfig,
-    /// App client
-    app_client: AppClient,
     /// Shutdown signal receiver
     rx_shutdown: broadcast::Receiver<()>,
 }
@@ -23,19 +25,16 @@ pub struct IndexerConfig {
     pub rpc_url: String,
     /// Bitcoin RPC user:password (optional)
     pub rpc_userpwd: Option<String>,
+    /// Path to the database storing the MMR accumulator
+    pub mmr_db_path: PathBuf,
     /// Indexing lag in blocks
     pub indexing_lag: u32,
 }
 
 impl Indexer {
-    pub fn new(
-        config: IndexerConfig,
-        app_client: AppClient,
-        rx_shutdown: broadcast::Receiver<()>,
-    ) -> Self {
+    pub fn new(config: IndexerConfig, rx_shutdown: broadcast::Receiver<()>) -> Self {
         Self {
             config,
-            app_client,
             rx_shutdown,
         }
     }
@@ -47,15 +46,24 @@ impl Indexer {
             BitcoinClient::new(self.config.rpc_url.clone(), self.config.rpc_userpwd.clone())?;
         info!("Bitcoin RPC client initialized");
 
-        let mut next_block_height = self.app_client.get_block_count().await?;
+        // We need to specify mmr_id to have deterministic keys in the database
+        let mmr_id = Some("blocks".to_string());
+        let store = Arc::new(
+            MMRStore::single_atomic_writer(&self.config.mmr_db_path, mmr_id.clone()).await?,
+        );
+        let hasher = Arc::new(StarkBlakeHasher::default());
+        let mut mmr = BlockMMR::new(store, hasher, mmr_id);
+        info!("MMR loaded from {}", self.config.mmr_db_path.display());
+
+        let mut next_block_height = mmr.get_block_count().await?;
         info!("Current MMR blocks count: {}", next_block_height);
 
         loop {
             tokio::select! {
                 res = bitcoin_client.wait_block_header(next_block_height, self.config.indexing_lag) => {
                     match res {
                         Ok((block_header, block_hash)) => {
-                            self.app_client.add_block(block_header).await?;
+                            mmr.add_block_header(next_block_height, &block_header).await?;
                             info!("Block #{} {} processed", next_block_height, block_hash);
                             next_block_height += 1;
                         },

crates/raito-bridge-node/src/main.rs

@@ -16,6 +16,7 @@ use crate::{
 
 mod app;
 mod indexer;
+mod mmr_store;
 mod rpc;
 mod shutdown;
 
@@ -67,29 +68,30 @@ async fn main() {
     // Instantiating components and wiring them together
     let shutdown = Shutdown::default();
 
+    let indexer_config = IndexerConfig {
+        rpc_url: cli.bitcoin_rpc_url.clone(),
+        rpc_userpwd: cli.bitcoin_rpc_userpwd.clone(),
+        mmr_db_path: cli.db_path.clone(),
+        indexing_lag: cli.mmr_block_lag,
+    };
+    let mut indexer = Indexer::new(indexer_config, shutdown.subscribe());
+
     let app_config = AppConfig {
-        db_path: cli.db_path,
+        db_path: cli.db_path.clone(),
         api_requests_capacity: 1000,
         bitcoin_rpc_url: cli.bitcoin_rpc_url.clone(),
         bitcoin_rpc_userpwd: cli.bitcoin_rpc_userpwd.clone(),
     };
     let (mut app_server, app_client) = create_app(app_config, shutdown.subscribe());
 
-    let indexer_config = IndexerConfig {
-        rpc_url: cli.bitcoin_rpc_url.clone(),
-        rpc_userpwd: cli.bitcoin_rpc_userpwd.clone(),
-        indexing_lag: cli.mmr_block_lag,
-    };
-    let mut indexer = Indexer::new(indexer_config, app_client.clone(), shutdown.subscribe());
-
     let rpc_config = RpcConfig {
         rpc_host: cli.rpc_host,
     };
     let rpc_server = RpcServer::new(rpc_config, app_client.clone(), shutdown.subscribe());
 
     // Launching threads for each component
-    let app_handle = tokio::spawn(async move { app_server.run().await });
     let indexer_handle = tokio::spawn(async move { indexer.run().await });
+    let app_handle = tokio::spawn(async move { app_server.run().await });
     let rpc_handle = tokio::spawn(async move { rpc_server.run().await });
     let shutdown_handle = tokio::spawn(async move { shutdown.run().await });