diff --git a/Cargo.toml b/Cargo.toml index a9455b37..4afd5766 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -6,9 +6,13 @@ members = ["crates/*"] cairo-vm = { git = "https://github.com/maciejka/cairo-vm", rev = "19d8a07ce9799a8af9db6f8a14a8accaad900214" } [workspace.dependencies] +# Accumulators +accumulators = { version = "0.5.1", 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"] } diff --git a/crates/raito-bridge-node/Cargo.toml b/crates/raito-bridge-node/Cargo.toml index 527aa88d..5e32033e 100644 --- a/crates/raito-bridge-node/Cargo.toml +++ b/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,6 +14,7 @@ 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"] } @@ -19,7 +22,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 diff --git a/crates/raito-bridge-node/src/app.rs b/crates/raito-bridge-node/src/app.rs index 3f670a99..9200681d 100644 --- a/crates/raito-bridge-node/src/app.rs +++ b/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::store::AppStore; + /// 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), - /// 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)), /// 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( + AppStore::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, diff --git a/crates/raito-bridge-node/src/indexer.rs b/crates/raito-bridge-node/src/indexer.rs index 159d2c5a..50d4262f 100644 --- a/crates/raito-bridge-node/src/indexer.rs +++ b/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::store::AppStore; /// 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, + /// 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,7 +46,16 @@ 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( + AppStore::single_atomic_writer(&self.config.mmr_db_path, mmr_id.clone()).await?, + ); + let hasher = Arc::new(StarkBlakeHasher::default()); + let mut mmr = BlockMMR::new(store.clone(), 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 { @@ -55,7 +63,9 @@ impl Indexer { 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?; + store.begin().await?; + mmr.add_block_header(next_block_height, &block_header).await?; + store.commit().await?; info!("Block #{} {} processed", next_block_height, block_hash); next_block_height += 1; }, diff --git a/crates/raito-bridge-node/src/main.rs b/crates/raito-bridge-node/src/main.rs index 37f99502..b839c9ef 100644 --- a/crates/raito-bridge-node/src/main.rs +++ b/crates/raito-bridge-node/src/main.rs @@ -18,6 +18,7 @@ mod app; mod indexer; mod rpc; mod shutdown; +mod store; #[derive(Parser)] #[command(author, version, about, long_about = None)] @@ -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 }); diff --git a/crates/raito-bridge-node/src/store.rs b/crates/raito-bridge-node/src/store.rs new file mode 100644 index 00000000..bf985b35 --- /dev/null +++ b/crates/raito-bridge-node/src/store.rs @@ -0,0 +1,186 @@ +use std::collections::HashMap; +use std::ops::DerefMut; +use std::path::Path; + +use accumulators::store::{sqlite::SQLiteStore, Store as AccumulatorsStore, StoreError}; +use async_trait::async_trait; +use bitcoin::block::Header as BlockHeader; +use bitcoin::consensus::{Decodable, Encodable}; +use raito_spv_mmr::block_mmr::BlockMMRStore; +use sqlx::sqlite::{ + SqliteConnectOptions, SqliteJournalMode, SqlitePoolOptions, SqliteSynchronous, + SqliteTransactionManager, +}; +use sqlx::{Row, TransactionManager}; +use tokio::fs; + +/// SQLite busy timeout in milliseconds +const SQLITE_BUSY_TIMEOUT: &str = "5000"; + +/// Maximum number of concurrent readers (size of the connection pool) +const SQLITE_MAX_CONCURRENT_READERS: u32 = 10; + +/// SQLite-backed store with single-writer and multi-reader pools. +/// - WAL mode for concurrent readers during writes +/// - Single writer (max_connections = 1) +/// - Optional active write transaction encapsulated in the store +#[derive(Debug)] +pub struct AppStore(SQLiteStore); + +impl AppStore { + /// Create a store for a single atomic writer + pub async fn single_atomic_writer>( + path: P, + id: Option, + ) -> Result { + if let Some(parent) = path.as_ref().parent() { + fs::create_dir_all(parent).await?; + } + + let options = SqliteConnectOptions::new() + .filename(path.as_ref()) + .create_if_missing(true) + .journal_mode(SqliteJournalMode::Wal) + .synchronous(SqliteSynchronous::Normal) + .pragma("busy_timeout", SQLITE_BUSY_TIMEOUT); + + // Writer pool: single connection ensures single-writer semantics + let pool = SqlitePoolOptions::new() + .max_connections(1) + .connect_with(options) + .await?; + + let store = Self(SQLiteStore::with_pool(pool, id)); + store.init().await?; + + Ok(store) + } + + /// Create a store for multiple concurrent readers + pub async fn multiple_concurrent_readers>( + path: P, + id: Option, + ) -> Result { + let options = SqliteConnectOptions::new() + .filename(path.as_ref()) + .read_only(true); + + let pool = SqlitePoolOptions::new() + .max_connections(SQLITE_MAX_CONCURRENT_READERS) + .connect_with(options) + .await?; + + Ok(Self(SQLiteStore::with_pool(pool, id))) + } + + /// Initialize the store by creating the tables if missing + async fn init(&self) -> Result<(), sqlx::Error> { + // Create a key-value store table for MMR accumulator state + self.0.init().await?; + // Create a table for encoded block headers + let mut conn = self.0.acquire_connection().await?; + sqlx::query( + r#"CREATE TABLE IF NOT EXISTS block_headers ( + height INTEGER PRIMARY KEY, + hash TEXT NOT NULL, + header BLOB NOT NULL + );"#, + ) + .execute(conn.deref_mut()) + .await?; + // Add index on block hash column + sqlx::query( + r#"CREATE INDEX IF NOT EXISTS idx_block_headers_hash ON block_headers (hash);"#, + ) + .execute(conn.deref_mut()) + .await?; + Ok(()) + } + + /// Begin a new transaction. + /// NOTE that this function does not check if there is already a transaction in progress. + pub async fn begin(&self) -> Result<(), StoreError> { + let mut conn = self.0.acquire_connection().await?; + SqliteTransactionManager::begin(&mut conn, None) + .await + .map_err(StoreError::SQLite) + } + + /// Commit the current transaction. + /// NOTE that this function does not check if there is a transaction in progress. + pub async fn commit(&self) -> Result<(), StoreError> { + let mut conn = self.0.acquire_connection().await?; + SqliteTransactionManager::commit(&mut conn) + .await + .map_err(StoreError::SQLite) + } +} + +#[async_trait] +impl BlockMMRStore for AppStore { + /// Add a new block header to the store + async fn add_block_header( + &self, + height: u32, + block_header: &BlockHeader, + ) -> Result<(), StoreError> { + let mut block_header_data = Vec::new(); + let mut conn = self.0.acquire_connection().await?; + block_header + .consensus_encode(&mut block_header_data) + .map_err(|e| StoreError::Custom(Box::new(e)))?; + sqlx::query("INSERT INTO block_headers (height, hash, header) VALUES (?, ?, ?)") + .bind(height) + .bind(block_header.block_hash().to_string()) + .bind(block_header_data) + .execute(conn.deref_mut()) + .await?; + Ok(()) + } + + /// Get a range of block headers from the store + async fn get_block_headers( + &self, + start_height: u32, + num_blocks: u32, + ) -> Result, StoreError> { + let mut conn = self.0.acquire_connection().await?; + let rows = sqlx::query("SELECT header FROM block_headers WHERE height >= ? AND height < ?") + .bind(start_height) + .bind(start_height + num_blocks) + .fetch_all(conn.deref_mut()) + .await?; + rows.iter() + .map(|row| { + let header: Vec = row.get("header"); + BlockHeader::consensus_decode(&mut header.as_slice()) + .map_err(|e| StoreError::Custom(Box::new(e))) + }) + .collect() + } +} + +#[async_trait] +impl AccumulatorsStore for AppStore { + fn id(&self) -> String { + self.0.id() + } + async fn get(&self, key: &str) -> Result, StoreError> { + self.0.get(key).await + } + async fn get_many(&self, keys: Vec<&str>) -> Result, StoreError> { + self.0.get_many(keys).await + } + async fn set(&self, key: &str, value: &str) -> Result<(), StoreError> { + self.0.set(key, value).await + } + async fn set_many(&self, entries: HashMap) -> Result<(), StoreError> { + self.0.set_many(entries).await + } + async fn delete(&self, key: &str) -> Result<(), StoreError> { + self.0.delete(key).await + } + async fn delete_many(&self, keys: Vec<&str>) -> Result<(), StoreError> { + self.0.delete_many(keys).await + } +} diff --git a/crates/raito-spv-mmr/Cargo.toml b/crates/raito-spv-mmr/Cargo.toml index 6bc932dd..d0d999fa 100644 --- a/crates/raito-spv-mmr/Cargo.toml +++ b/crates/raito-spv-mmr/Cargo.toml @@ -7,8 +7,10 @@ edition = "2021" sqlite = ["accumulators/sqlite"] [dependencies] +async-trait.workspace = true + # Merkle mountain range -accumulators = { git = "https://github.com/maciejka/rust-accumulators", rev = "1fbc79a472f1659f0aa0213de95a914086732204", features = ["blake", "memory", "mmr"]} +accumulators.workspace = true # Bitcoin RPC and types #jsonrpsee.workspace = true @@ -42,4 +44,4 @@ tokio = { version = "1.36", default-features = false, features = ["rt", "macros" [dev-dependencies] # Testing mockall.workspace = true -tempfile.workspace = true \ No newline at end of file +tempfile.workspace = true \ No newline at end of file diff --git a/crates/raito-spv-mmr/src/block_mmr.rs b/crates/raito-spv-mmr/src/block_mmr.rs index fee7bb50..d8587e60 100644 --- a/crates/raito-spv-mmr/src/block_mmr.rs +++ b/crates/raito-spv-mmr/src/block_mmr.rs @@ -1,9 +1,6 @@ //! Merkle Mountain Range (MMR) accumulator implementation for Bitcoin block headers with proof generation. -use std::path::Path; use std::sync::Arc; -#[cfg(feature = "sqlite")] -use tokio::fs; use accumulators::hasher::stark_blake::StarkBlakeHasher; use accumulators::hasher::Hasher; @@ -12,9 +9,8 @@ use accumulators::mmr::{ PeaksOptions, Proof, ProofOptions, MMR, }; use accumulators::store::memory::InMemoryStore; -#[cfg(feature = "sqlite")] -use accumulators::store::sqlite::SQLiteStore; -use accumulators::store::Store; +use accumulators::store::{Store, StoreError}; +use async_trait::async_trait; use bitcoin::block::Header as BlockHeader; use bitcoin::hashes::Hash; use serde::{Deserialize, Serialize}; @@ -25,11 +21,35 @@ use crate::sparse_roots::SparseRoots; #[derive(Debug)] pub struct BlockMMR { hasher: Arc, - #[allow(dead_code)] - store: Arc, + store: Arc, mmr: MMR, } +#[async_trait] +pub trait BlockMMRStore: Store { + async fn add_block_header( + &self, + height: u32, + block_header: &BlockHeader, + ) -> Result<(), StoreError>; + async fn get_block_headers( + &self, + start_height: u32, + num_blocks: u32, + ) -> Result, StoreError>; +} + +#[async_trait] +impl BlockMMRStore for InMemoryStore { + async fn add_block_header(&self, _: u32, _: &BlockHeader) -> Result<(), StoreError> { + tracing::warn!("Adding block header to in-memory store is not supported"); + Ok(()) + } + async fn get_block_headers(&self, _: u32, _: u32) -> Result, StoreError> { + unimplemented!("Getting block headers from in-memory store is not supported"); + } +} + /// Proof data structure for demonstrating inclusion of a block in the MMR #[derive(Debug, Clone, Serialize, Deserialize)] pub struct BlockInclusionProof { @@ -54,30 +74,13 @@ impl Default for BlockMMR { impl BlockMMR { /// Create a new default MMR - pub fn new(store: Arc, hasher: Arc, mmr_id: Option) -> Self { + pub fn new( + store: Arc, + hasher: Arc, + mmr_id: Option, + ) -> Self { let mmr = MMR::new(store.clone(), hasher.clone(), mmr_id); - Self { hasher, store, mmr } - } - - /// Create MMR from file - #[cfg(feature = "sqlite")] - pub async fn from_file(path: &Path, mmr_id: &str) -> Result { - if let Some(parent) = path.parent() { - fs::create_dir_all(parent).await?; - } - - let store = - Arc::new(SQLiteStore::new(path.to_str().unwrap(), Some(true), Some(mmr_id)).await?); - let hasher = Arc::new(StarkBlakeHasher::default()); - Ok(Self::new(store, hasher, Some(mmr_id.to_string()))) - } - - /// Create MMR from file - #[cfg(not(feature = "sqlite"))] - pub async fn from_file(_path: &Path, _mmr_id: &str) -> Result { - Err(anyhow::anyhow!( - "SQLite support is disabled. Enable the 'sqlite' feature to use this method." - )) + Self { hasher, mmr, store } } /// Create in-memory MMR from peaks hashes and elements count @@ -95,7 +98,7 @@ impl BlockMMR { leaf_count_to_mmr_size(leaf_count), ) .await?; - Ok(Self { hasher, store, mmr }) + Ok(Self { hasher, mmr, store }) } /// Add a leaf to the MMR @@ -105,9 +108,39 @@ impl BlockMMR { } /// Add a block header to the MMR - pub async fn add_block_header(&mut self, block_header: &BlockHeader) -> anyhow::Result<()> { + pub async fn add_block_header( + &mut self, + height: u32, + block_header: &BlockHeader, + ) -> anyhow::Result<()> { let leaf = block_header_digest(self.hasher.clone(), block_header)?; - self.add(leaf).await + self.add(leaf).await?; + self.store + .add_block_header(height, block_header) + .await + .map_err(|e| anyhow::anyhow!("Failed to add block header: {}", e))?; + Ok(()) + } + + /// Get a range of block headers from the MMR + pub async fn get_block_headers( + &self, + start_height: u32, + num_blocks: u32, + ) -> anyhow::Result> { + let res = self + .store + .get_block_headers(start_height, num_blocks) + .await + .map_err(|e| anyhow::anyhow!("Failed to get block headers: {}", e))?; + if res.len() != num_blocks as usize { + return Err(anyhow::anyhow!( + "Failed to get block headers: expected {}, got {}", + num_blocks, + res.len() + )); + } + Ok(res) } /// Get the number of blocks in the MMR (number of leaves) @@ -417,8 +450,8 @@ mod tests { ) .unwrap(); // Add 10 blocks - for _ in 0..10 { - mmr.add_block_header(&block_header).await.unwrap(); + for i in 0..10 { + mmr.add_block_header(i as u32, &block_header).await.unwrap(); } // Generate a proof for the fifth block let proof = mmr.generate_proof(5, None).await.unwrap();