diff --git a/Cargo.toml b/Cargo.toml index 4afd5766..ac34763c 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -36,6 +36,7 @@ blake2 = "0.10" serde = { version = "1.0", features = ["derive"] } serde_json = { version = "1.0", features = ["arbitrary_precision"] } hex = "0.4" +bincode = "1.3" # BigInt num-bigint = { version = "0.4", features = ["serde"] } diff --git a/crates/raito-bridge-node/Cargo.toml b/crates/raito-bridge-node/Cargo.toml index 637a08dd..ddc196b6 100644 --- a/crates/raito-bridge-node/Cargo.toml +++ b/crates/raito-bridge-node/Cargo.toml @@ -22,6 +22,7 @@ tower-http = { version = "0.5", features = ["trace", "cors", "compression-gzip"] bitcoin.workspace = true # Storage sqlx.workspace = true +bincode.workspace = true # CLI clap.workspace = true dotenv.workspace = true diff --git a/crates/raito-bridge-node/src/chain_state.rs b/crates/raito-bridge-node/src/chain_state.rs new file mode 100644 index 00000000..c894dec9 --- /dev/null +++ b/crates/raito-bridge-node/src/chain_state.rs @@ -0,0 +1,111 @@ +use std::str::FromStr; +use std::sync::Arc; + +use accumulators::store::StoreError; +use async_trait::async_trait; +use bitcoin::block::BlockHash; +use bitcoin::block::Header as BlockHeader; +use bitcoin::Target; +use bitcoin::Work; +use raito_spv_verify::ChainState; + +const BLOCKS_PER_EPOCH: u32 = 2016; + +#[async_trait] +pub trait ChainStateStore: Send + Sync { + 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 fn get_block_height(&self, block_hash: &BlockHash) -> Result; + async fn add_chain_state( + &self, + height: u32, + chain_state: &ChainState, + ) -> Result<(), StoreError>; + async fn get_chain_state(&self, height: u32) -> Result; +} + +pub struct ChainStateManager { + current_state: ChainState, + store: Arc, +} + +impl ChainStateManager { + pub async fn restore( + store: Arc, + height: u32, + ) -> Result { + let current_state = if height == 0 { + Self::genesis_state() + } else { + store.get_chain_state(height - 1).await? + }; + Ok(Self { + current_state, + store, + }) + } + + pub async fn update( + &mut self, + block_height: u32, + block_header: &BlockHeader, + ) -> Result<(), anyhow::Error> { + let new_state = if block_height == 0 { + self.current_state.clone() + } else { + let mut prev_timestamps = self.current_state.prev_timestamps.clone(); + prev_timestamps.push(block_header.time); + if prev_timestamps.len() > 11 { + prev_timestamps.remove(0); + } + + let epoch_start_time = if block_height % BLOCKS_PER_EPOCH == 0 { + block_header.time + } else { + self.current_state.epoch_start_time + }; + + ChainState { + block_height, + total_work: self.current_state.total_work + block_header.work(), + best_block_hash: block_header.block_hash(), + current_target: block_header.target(), + epoch_start_time, + prev_timestamps, + } + }; + + self.store.add_chain_state(block_height, &new_state).await?; + self.store + .add_block_header(block_height, block_header) + .await?; + self.current_state = new_state; + + Ok(()) + } + + pub fn genesis_state() -> ChainState { + ChainState { + block_height: 0, + total_work: Work::from_hex("0x100010001").unwrap(), + best_block_hash: BlockHash::from_str( + "000000000019d6689c085ae165831e934ff763ae46a2a6c172b3f1b60a8ce26f", + ) + .unwrap(), + current_target: Target::from_hex( + "0xffff0000000000000000000000000000000000000000000000000000", + ) + .unwrap(), + epoch_start_time: 1231006505, + prev_timestamps: vec![1231006505], + } + } +} diff --git a/crates/raito-bridge-node/src/indexer.rs b/crates/raito-bridge-node/src/indexer.rs index a8109d65..373d3045 100644 --- a/crates/raito-bridge-node/src/indexer.rs +++ b/crates/raito-bridge-node/src/indexer.rs @@ -9,7 +9,7 @@ use tracing::{error, info}; use raito_bitcoin_client::BitcoinClient; -use crate::store::AppStore; +use crate::{chain_state::ChainStateManager, store::AppStore}; /// Bitcoin block indexer that builds MMR accumulator and generates sparse roots pub struct Indexer { @@ -60,13 +60,18 @@ impl Indexer { let mut next_block_height = mmr.get_block_count().await?; info!("Current MMR blocks count: {}", next_block_height); + let mut chain_state_mgr = + ChainStateManager::restore(store.clone(), next_block_height).await?; + info!("Chain state manager initialized"); + loop { tokio::select! { res = bitcoin_client.wait_block_header(next_block_height, self.config.indexing_lag) => { match res { Ok((block_header, block_hash)) => { store.begin().await?; - mmr.add_block_header(next_block_height, &block_header).await?; + mmr.add_block_header(&block_header).await.map_err(|e| anyhow::anyhow!("Failed to add block header to MMR: {}", e))?; + chain_state_mgr.update(next_block_height, &block_header).await.map_err(|e| anyhow::anyhow!("Failed to update chain state: {}", e))?; 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 34189e28..eda8705e 100644 --- a/crates/raito-bridge-node/src/main.rs +++ b/crates/raito-bridge-node/src/main.rs @@ -13,6 +13,7 @@ use crate::{ shutdown::Shutdown, }; +mod chain_state; mod indexer; mod rpc; mod shutdown; diff --git a/crates/raito-bridge-node/src/rpc.rs b/crates/raito-bridge-node/src/rpc.rs index f6015f97..922cec1b 100644 --- a/crates/raito-bridge-node/src/rpc.rs +++ b/crates/raito-bridge-node/src/rpc.rs @@ -21,9 +21,9 @@ use raito_spv_mmr::{ block_mmr::{BlockInclusionProof, BlockMMR}, sparse_roots::SparseRoots, }; -use raito_spv_verify::TransactionInclusionProof; +use raito_spv_verify::{ChainState, TransactionInclusionProof}; -use crate::store::AppStore; +use crate::{chain_state::ChainStateStore, store::AppStore}; /// Query parameters for block inclusion proof generation and roots retrieval #[derive(Debug, Deserialize)] @@ -62,21 +62,25 @@ pub struct RpcServer { #[derive(Debug, Clone)] pub struct AppState { mmr: Arc, + store: Arc, bitcoin_client: Arc, } impl AppState { - pub async fn new(config: RpcConfig) -> Result { + pub fn new(config: RpcConfig) -> Result { let mmr_id = Some(config.mmr_id.clone()); - let store = - AppStore::multiple_concurrent_readers(&config.mmr_db_path, mmr_id.clone()).await?; + let store = Arc::new(AppStore::multiple_concurrent_readers( + &config.mmr_db_path, + mmr_id.clone(), + )); let hasher = StarkBlakeHasher::default(); - let mmr = BlockMMR::new(Arc::new(store), Arc::new(hasher), mmr_id); + let mmr = BlockMMR::new(store.clone(), Arc::new(hasher), mmr_id); let bitcoin_client = BitcoinClient::new(config.rpc_url.clone(), config.rpc_userpwd.clone())?; Ok(Self { mmr: Arc::new(mmr), bitcoin_client: Arc::new(bitcoin_client), + store: store.clone(), }) } } @@ -93,7 +97,6 @@ impl RpcServer { info!("Starting RPC server on {}", self.config.rpc_host); let app_state = AppState::new(self.config.clone()) - .await .map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, e))?; let app = Router::new() @@ -103,6 +106,7 @@ impl RpcServer { .route("/headers", get(get_block_headers)) .route("/transaction-proof/:tx_id", get(get_transaction_proof)) .route("/block-header/:block_height", get(get_block_header)) + .route("/chain-state/:block_height", get(get_chain_state)) .with_state(app_state) .layer(CompressionLayer::new()) .layer(CorsLayer::permissive()) @@ -210,7 +214,7 @@ pub async fn get_block_header( Path(block_height): Path, ) -> Result, StatusCode> { let block_header = state - .mmr + .store .get_block_headers(block_height, 1) .await .map_err(|e| { @@ -220,9 +224,8 @@ pub async fn get_block_header( ); StatusCode::INTERNAL_SERVER_ERROR })? - .get(0) - .ok_or(StatusCode::INTERNAL_SERVER_ERROR)? - .clone(); + .pop() + .ok_or(StatusCode::INTERNAL_SERVER_ERROR)?; Ok(Json(block_header)) } @@ -242,7 +245,7 @@ pub async fn get_block_headers( let offset = query.offset.unwrap_or(0); let size = query.size.unwrap_or(10); let block_headers = state - .mmr + .store .get_block_headers(offset, size) .await .map_err(|e| { @@ -282,13 +285,17 @@ pub async fn get_transaction_proof( })?; let block_hash = block_header.block_hash(); - let block_height = state.mmr.get_block_height(&block_hash).await.map_err(|e| { - error!( - "Failed to get block height for block hash {}: {}", - block_hash, e - ); - StatusCode::INTERNAL_SERVER_ERROR - })?; + let block_height = state + .store + .get_block_height(&block_hash) + .await + .map_err(|e| { + error!( + "Failed to get block height for block hash {}: {}", + block_hash, e + ); + StatusCode::INTERNAL_SERVER_ERROR + })?; let transaction = state .bitcoin_client @@ -307,3 +314,26 @@ pub async fn get_transaction_proof( }; Ok(Json(transaction_proof.into())) } + +/// Get the chain state for a specific block height +/// +/// # Returns +/// * `Json` - The chain state in JSON format +/// * `StatusCode::INTERNAL_SERVER_ERROR` - If fetching the chain state fails +pub async fn get_chain_state( + State(state): State, + Path(block_height): Path, +) -> Result, StatusCode> { + let chain_state = state + .store + .get_chain_state(block_height) + .await + .map_err(|e| { + error!( + "Failed to get chain state for height {}: {}", + block_height, e + ); + StatusCode::INTERNAL_SERVER_ERROR + })?; + Ok(Json(chain_state)) +} diff --git a/crates/raito-bridge-node/src/store.rs b/crates/raito-bridge-node/src/store.rs index 1ff21c3b..a11a85b5 100644 --- a/crates/raito-bridge-node/src/store.rs +++ b/crates/raito-bridge-node/src/store.rs @@ -7,7 +7,7 @@ use async_trait::async_trait; use bitcoin::block::Header as BlockHeader; use bitcoin::consensus::{Decodable, Encodable}; use bitcoin::BlockHash; -use raito_spv_mmr::block_mmr::BlockMMRStore; +use raito_spv_verify::ChainState; use sqlx::sqlite::{ SqliteConnectOptions, SqliteJournalMode, SqlitePoolOptions, SqliteSynchronous, SqliteTransactionManager, @@ -15,6 +15,8 @@ use sqlx::sqlite::{ use sqlx::{Row, TransactionManager}; use tokio::fs; +use crate::chain_state::ChainStateStore; + /// SQLite busy timeout in milliseconds const SQLITE_BUSY_TIMEOUT: &str = "5000"; @@ -58,20 +60,16 @@ impl AppStore { } /// Create a store for multiple concurrent readers - pub async fn multiple_concurrent_readers>( - path: P, - id: Option, - ) -> Result { + pub fn multiple_concurrent_readers>(path: P, id: Option) -> Self { 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?; + .connect_lazy_with(options); - Ok(Self(SQLiteStore::with_pool(pool, id))) + Self(SQLiteStore::with_pool(pool, id)) } /// Initialize the store by creating the tables if missing @@ -89,6 +87,15 @@ impl AppStore { ) .execute(conn.deref_mut()) .await?; + // Create a table for chain states + sqlx::query( + r#"CREATE TABLE IF NOT EXISTS chain_states ( + height INTEGER PRIMARY KEY, + state 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);"#, @@ -118,7 +125,7 @@ impl AppStore { } #[async_trait] -impl BlockMMRStore for AppStore { +impl ChainStateStore for AppStore { /// Add a new block header to the store async fn add_block_header( &self, @@ -169,6 +176,31 @@ impl BlockMMRStore for AppStore { .await?; row.map(|row| row.get("height")).ok_or(StoreError::GetError) } + + async fn get_chain_state(&self, height: u32) -> Result { + let mut conn = self.0.acquire_connection().await?; + let row = sqlx::query("SELECT state FROM chain_states WHERE height = ?") + .bind(height) + .fetch_optional(conn.deref_mut()) + .await?; + let data: Vec = row.ok_or(StoreError::GetError)?.get("state"); + bincode::deserialize::(&data).map_err(|e| StoreError::Custom(Box::new(e))) + } + + async fn add_chain_state( + &self, + height: u32, + chain_state: &ChainState, + ) -> Result<(), StoreError> { + let mut conn = self.0.acquire_connection().await?; + let data = bincode::serialize(chain_state).map_err(|e| StoreError::Custom(Box::new(e)))?; + sqlx::query("INSERT INTO chain_states (height, state) VALUES (?, ?)") + .bind(height) + .bind(data) + .execute(conn.deref_mut()) + .await?; + Ok(()) + } } #[async_trait] diff --git a/crates/raito-spv-client/Cargo.toml b/crates/raito-spv-client/Cargo.toml index 8310b49b..2c27de6d 100644 --- a/crates/raito-spv-client/Cargo.toml +++ b/crates/raito-spv-client/Cargo.toml @@ -41,7 +41,7 @@ starknet-ff = "0.3.7" hex = "0.4.3" serde = { workspace = true } serde_json = { workspace = true } -bincode = "1.3" +bincode = { workspace = true } # Compression bzip2 = "0.4" diff --git a/crates/raito-spv-mmr/src/block_mmr.rs b/crates/raito-spv-mmr/src/block_mmr.rs index b3710fc9..4fdd122d 100644 --- a/crates/raito-spv-mmr/src/block_mmr.rs +++ b/crates/raito-spv-mmr/src/block_mmr.rs @@ -9,11 +9,9 @@ use accumulators::mmr::{ PeaksOptions, Proof, ProofOptions, MMR, }; use accumulators::store::memory::InMemoryStore; -use accumulators::store::{Store, StoreError}; -use async_trait::async_trait; +use accumulators::store::Store; use bitcoin::block::Header as BlockHeader; use bitcoin::hashes::Hash; -use bitcoin::BlockHash; use serde::{Deserialize, Serialize}; use crate::sparse_roots::SparseRoots; @@ -22,39 +20,9 @@ use crate::sparse_roots::SparseRoots; #[derive(Debug)] pub struct BlockMMR { hasher: 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 fn get_block_height(&self, block_hash: &BlockHash) -> Result; -} - -#[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"); - } - async fn get_block_height(&self, _: &BlockHash) -> Result { - unimplemented!("Getting block height 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 { @@ -79,13 +47,9 @@ 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, mmr, store } + Self { hasher, mmr } } /// Create in-memory MMR from peaks hashes and elements count @@ -103,7 +67,7 @@ impl BlockMMR { leaf_count_to_mmr_size(leaf_count), ) .await?; - Ok(Self { hasher, mmr, store }) + Ok(Self { hasher, mmr }) } /// Add a leaf to the MMR @@ -113,49 +77,12 @@ impl BlockMMR { } /// Add a block header to the MMR - pub async fn add_block_header( - &mut self, - height: u32, - block_header: &BlockHeader, - ) -> anyhow::Result<()> { + pub async fn add_block_header(&mut self, block_header: &BlockHeader) -> anyhow::Result<()> { let leaf = block_header_digest(self.hasher.clone(), block_header)?; 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 height of a block by its hash - pub async fn get_block_height(&self, block_hash: &BlockHash) -> anyhow::Result { - self.store - .get_block_height(block_hash) - .await - .map_err(|e| anyhow::anyhow!("Failed to get block height: {}", e)) - } - /// Get the number of blocks in the MMR (number of leaves) pub async fn get_block_count(&self) -> anyhow::Result { self.mmr @@ -463,8 +390,8 @@ mod tests { ) .unwrap(); // Add 10 blocks - for i in 0..10 { - mmr.add_block_header(i as u32, &block_header).await.unwrap(); + for _ in 0..10 { + mmr.add_block_header(&block_header).await.unwrap(); } // Generate a proof for the fifth block let proof = mmr.generate_proof(5, None).await.unwrap(); diff --git a/crates/raito-spv-verify/src/proof.rs b/crates/raito-spv-verify/src/proof.rs index 3070218b..c5f3dcfe 100644 --- a/crates/raito-spv-verify/src/proof.rs +++ b/crates/raito-spv-verify/src/proof.rs @@ -1,12 +1,10 @@ //! Types representing the compressed SPV proof and helpers to decode Cairo outputs //! and compute chain state digests used during verification. -use std::str::FromStr; - use bitcoin::hashes::Hash; use bitcoin::{block::Header as BlockHeader, BlockHash, Transaction}; +use bitcoin::{Target, Work}; use cairo_air::CairoProof; -use num_bigint::BigUint; use raito_spv_mmr::block_mmr::BlockInclusionProof; use serde::{Deserialize, Serialize}; use starknet_ff::FieldElement; @@ -45,16 +43,16 @@ pub struct CompressedSpvProof { } /// Snapshot of the consensus chain state used to validate block inclusion -#[derive(Debug, Serialize, Deserialize)] +#[derive(Debug, Clone, Serialize, Deserialize)] pub struct ChainState { /// The height of the best block in the chain pub block_height: u32, - /// The total accumulated work of the chain as a decimal string - pub total_work: String, + /// The total accumulated work of the chain + pub total_work: Work, /// The hash of the best block in the chain pub best_block_hash: BlockHash, - /// The current target difficulty as a compact decimal string - pub current_target: String, + /// The current target difficulty + pub current_target: Target, /// The start time (UNIX seconds) of the current difficulty epoch pub epoch_start_time: u32, /// The timestamps (UNIX seconds) of the previous 11 blocks @@ -158,9 +156,9 @@ impl ChainState { // Construct the payload for the hash function, all integers are little-endian let mut words = Vec::new(); words.push(self.block_height); - words.extend_from_slice(&big_uint_to_u256_words(&self.total_work)?); + words.extend_from_slice(&split_bytes_into_words(&self.total_work.to_be_bytes())); words.extend_from_slice(&best_block_hash_words); - words.extend_from_slice(&big_uint_to_u256_words(&self.current_target)?); + words.extend_from_slice(&split_bytes_into_words(&self.current_target.to_be_bytes())); words.push(self.epoch_start_time); words.extend_from_slice(&self.prev_timestamps); @@ -184,12 +182,11 @@ impl ChainState { } } -fn big_uint_to_u256_words(value: &str) -> Result, anyhow::Error> { - let number = BigUint::from_str(value).map_err(|_| anyhow::anyhow!("Invalid number"))?; - let mut digits = number.to_u32_digits(); - digits.extend(vec![0; 8 - digits.len()]); - digits.reverse(); - Ok(digits) +fn split_bytes_into_words(bytes: &[u8]) -> Vec { + bytes + .chunks_exact(4) + .map(|chunk| u32::from_be_bytes(chunk.try_into().unwrap())) + .collect() } #[cfg(test)] @@ -202,13 +199,15 @@ mod tests { fn test_chain_state_hash() { let chain_state = ChainState { block_height: 0, - total_work: "4295032833".to_string(), + total_work: Work::from_hex("0x100010001").unwrap(), best_block_hash: BlockHash::from_str( "000000000019d6689c085ae165831e934ff763ae46a2a6c172b3f1b60a8ce26f", ) .unwrap(), - current_target: "26959535291011309493156476344723991336010898738574164086137773096960" - .to_string(), + current_target: Target::from_hex( + "0xffff0000000000000000000000000000000000000000000000000000", + ) + .unwrap(), epoch_start_time: 1231006505, prev_timestamps: vec![1231006505], }; diff --git a/crates/raito-spv-verify/src/work.rs b/crates/raito-spv-verify/src/work.rs index f111ef25..29228a5b 100644 --- a/crates/raito-spv-verify/src/work.rs +++ b/crates/raito-spv-verify/src/work.rs @@ -20,13 +20,13 @@ pub fn verify_subchain_work( let start_epoch = chain_state.block_height / 2016; let end_epoch = block_height / 2016; let mut subchain_work = BigUint::ZERO; - let mut target = BigUint::from_str(&chain_state.current_target).unwrap(); + let mut target = BigUint::from_bytes_be(&chain_state.current_target.to_be_bytes()); for epoch in (end_epoch..=start_epoch).rev() { let start_block = min(2016 * (epoch + 1), chain_state.block_height); let end_block = max(2016 * epoch, block_height); let block_span = BigUint::from(start_block - end_block); - let block_work = compute_work_from_target(target.clone()); + let block_work = compute_work_from_target(&target); subchain_work += block_work * block_span; target *= BigUint::from(4_u32); } @@ -48,7 +48,7 @@ pub fn verify_subchain_work( } /// Compute the expected work for a single block given the target difficulty. -fn compute_work_from_target(target: BigUint) -> BigUint { +fn compute_work_from_target(target: &BigUint) -> BigUint { // 2^256 let max_work = BigUint::from_str( "115792089237316195423570985008687907853269984665640564039457584007913129639936",