diff --git a/dash-spv/src/storage/blocks.rs b/dash-spv/src/storage/blocks.rs index 430cb17e4..7d17f45dd 100644 --- a/dash-spv/src/storage/blocks.rs +++ b/dash-spv/src/storage/blocks.rs @@ -10,10 +10,9 @@ use dashcore::BlockHash; use tokio::sync::RwLock; use crate::error::StorageResult; -use crate::storage::io::atomic_write; use crate::storage::segments::SegmentCache; use crate::storage::PersistentStorage; -use crate::StorageError; +use crate::types::HashedBlockHeader; #[async_trait] pub trait BlockHeaderStorage { @@ -60,13 +59,12 @@ pub trait BlockHeaderStorage { } pub struct PersistentBlockHeaderStorage { - block_headers: RwLock>, + block_headers: RwLock>, header_hash_index: HashMap, } impl PersistentBlockHeaderStorage { const FOLDER_NAME: &str = "block_headers"; - const INDEX_FILE_NAME: &str = "index.dat"; } #[async_trait] @@ -75,24 +73,19 @@ impl PersistentStorage for PersistentBlockHeaderStorage { let storage_path = storage_path.into(); let segments_folder = storage_path.join(Self::FOLDER_NAME); - let index_path = segments_folder.join(Self::INDEX_FILE_NAME); + let mut block_headers: SegmentCache = + SegmentCache::load_or_new(&segments_folder).await?; - let mut block_headers = SegmentCache::load_or_new(&segments_folder).await?; + let mut header_hash_index = HashMap::new(); - let header_hash_index = match tokio::fs::read(&index_path) - .await - .ok() - .and_then(|content| bincode::deserialize(&content).ok()) + if let (Some(start), Some(end)) = (block_headers.start_height(), block_headers.tip_height()) { - Some(index) => index, - _ => { - if segments_folder.exists() { - block_headers.build_block_index_from_segments().await? - } else { - HashMap::new() - } + let headers = block_headers.get_items(start..end + 1).await?; + for (i, header) in headers.iter().enumerate() { + let height = start + i as u32; + header_hash_index.insert(*header.hash(), height); } - }; + } Ok(Self { block_headers: RwLock::new(block_headers), @@ -102,16 +95,12 @@ impl PersistentStorage for PersistentBlockHeaderStorage { async fn persist(&mut self, storage_path: impl Into + Send) -> StorageResult<()> { let block_headers_folder = storage_path.into().join(Self::FOLDER_NAME); - let index_path = block_headers_folder.join(Self::INDEX_FILE_NAME); tokio::fs::create_dir_all(&block_headers_folder).await?; self.block_headers.write().await.persist(&block_headers_folder).await; - let data = bincode::serialize(&self.header_hash_index) - .map_err(|e| StorageError::WriteFailed(format!("Failed to serialize index: {}", e)))?; - - atomic_write(&index_path, &data).await + Ok(()) } } @@ -128,13 +117,13 @@ impl BlockHeaderStorage for PersistentBlockHeaderStorage { height: u32, ) -> StorageResult<()> { let mut height = height; + let headers = + headers.iter().map(HashedBlockHeader::from).collect::>(); - let hashes = headers.iter().map(|header| header.block_hash()).collect::>(); - - self.block_headers.write().await.store_items_at_height(headers, height).await?; + self.block_headers.write().await.store_items_at_height(&headers, height).await?; - for hash in hashes { - self.header_hash_index.insert(hash, height); + for header in headers { + self.header_hash_index.insert(*header.hash(), height); height += 1; } @@ -142,7 +131,15 @@ impl BlockHeaderStorage for PersistentBlockHeaderStorage { } async fn load_headers(&self, range: Range) -> StorageResult> { - self.block_headers.write().await.get_items(range).await + Ok(self + .block_headers + .write() + .await + .get_items(range) + .await? + .into_iter() + .map(|cached| *cached.header()) + .collect()) } async fn get_tip_height(&self) -> Option { diff --git a/dash-spv/src/storage/segments.rs b/dash-spv/src/storage/segments.rs index 9401ab722..905b63cbf 100644 --- a/dash-spv/src/storage/segments.rs +++ b/dash-spv/src/storage/segments.rs @@ -17,7 +17,9 @@ use dashcore::{ }; use dashcore_hashes::Hash; -use crate::{error::StorageResult, storage::io::atomic_write, StorageError}; +use crate::{ + error::StorageResult, storage::io::atomic_write, types::HashedBlockHeader, StorageError, +}; pub trait Persistable: Sized + Encodable + Decodable + PartialEq + Clone { const SEGMENT_PREFIX: &'static str = "segment"; @@ -36,16 +38,18 @@ impl Persistable for Vec { } } -impl Persistable for BlockHeader { +impl Persistable for HashedBlockHeader { fn sentinel() -> Self { - Self { + let header = BlockHeader { version: Version::from_consensus(i32::MAX), // Invalid version prev_blockhash: BlockHash::from_byte_array([0xFF; 32]), // All 0xFF pattern merkle_root: dashcore::hashes::sha256d::Hash::from_byte_array([0xFF; 32]).into(), time: u32::MAX, // Far future timestamp bits: CompactTarget::from_consensus(0xFFFFFFFF), // Invalid difficulty nonce: u32::MAX, - } + }; + + Self::from(header) } } @@ -65,48 +69,6 @@ pub struct SegmentCache { segments_dir: PathBuf, } -impl SegmentCache { - pub async fn build_block_index_from_segments( - &mut self, - ) -> StorageResult> { - let entries = fs::read_dir(&self.segments_dir)?; - - let mut block_index = HashMap::new(); - - for entry in entries.flatten() { - let name = match entry.file_name().into_string() { - Ok(s) => s, - Err(_) => continue, - }; - - if !name.starts_with(BlockHeader::SEGMENT_PREFIX) { - continue; - } - - if !name.ends_with(&format!(".{}", BlockHeader::DATA_FILE_EXTENSION)) { - continue; - } - - let segment_id = match name[8..12].parse::() { - Ok(id) => id, - Err(_) => continue, - }; - - let mut block_height = Self::segment_id_to_start_height(segment_id); - - let segment = self.get_segment(&segment_id).await?; - - for item in segment.items.iter() { - block_index.insert(item.block_hash(), block_height); - - block_height += 1; - } - } - - Ok(block_index) - } -} - impl SegmentCache { const MAX_ACTIVE_SEGMENTS: usize = 10; diff --git a/dash-spv/src/sync/headers/manager.rs b/dash-spv/src/sync/headers/manager.rs index 1faf92c37..00a0a0c48 100644 --- a/dash-spv/src/sync/headers/manager.rs +++ b/dash-spv/src/sync/headers/manager.rs @@ -14,7 +14,7 @@ use crate::network::NetworkManager; use crate::storage::StorageManager; use crate::sync::headers::validate_headers; use crate::sync::headers2::Headers2StateManager; -use crate::types::{CachedHeader, ChainState}; +use crate::types::{ChainState, HashedBlockHeader}; use crate::ValidationMode; use std::sync::Arc; use tokio::sync::RwLock; @@ -133,8 +133,7 @@ impl HeaderSyncManager { // Wrap headers in CachedHeader to avoid redundant X11 hashing // This prevents recomputing hashes during validation, logging, and storage - let cached_headers: Vec = - headers.iter().map(|h| CachedHeader::new(*h)).collect(); + let cached_headers: Vec<_> = headers.iter().map(HashedBlockHeader::from).collect(); // Step 2: Validate Batch let first_cached = &cached_headers[0]; @@ -154,10 +153,10 @@ impl HeaderSyncManager { // Check if the first header connects to our tip // Cache tip hash to avoid recomputing it - let tip_cached = CachedHeader::new(tip); - let tip_hash = tip_cached.block_hash(); + let tip_cached = HashedBlockHeader::from(tip); + let tip_hash = tip_cached.hash(); - if first_header.prev_blockhash != tip_hash { + if first_header.prev_blockhash != *tip_hash { tracing::warn!( "Received header batch that does not connect to our tip. Expected prev_hash: {}, got: {}. Dropping message.", tip_hash, @@ -201,8 +200,8 @@ impl HeaderSyncManager { if !cached_headers.is_empty() { let last_cached = cached_headers.last().unwrap(); // Use cached hashes to avoid redundant X11 computation - let first_hash = first_cached.block_hash(); - let last_hash = last_cached.block_hash(); + let first_hash = first_cached.hash(); + let last_hash = last_cached.hash(); tracing::debug!( "Received headers batch: first.prev_hash={}, first.hash={}, last.hash={}, count={}", first_header.prev_blockhash, @@ -220,8 +219,8 @@ impl HeaderSyncManager { if self.reorg_config.enforce_checkpoints { // Use cached hash to avoid redundant X11 computation in loop - let header_hash = cached_header.block_hash(); - if !self.checkpoint_manager.validate_block(prospective_height, &header_hash) { + let header_hash = cached_header.hash(); + if !self.checkpoint_manager.validate_block(prospective_height, header_hash) { return Err(SyncError::Validation(format!( "Block at height {} does not match checkpoint", prospective_height @@ -259,8 +258,8 @@ impl HeaderSyncManager { // During sync mode - request next batch // Use the last cached header's hash to avoid redundant X11 computation if let Some(last_cached) = cached_headers.last() { - let hash = last_cached.block_hash(); - self.request_headers(network, Some(hash), storage).await?; + let hash = last_cached.hash(); + self.request_headers(network, Some(*hash), storage).await?; } } diff --git a/dash-spv/src/sync/headers/validation.rs b/dash-spv/src/sync/headers/validation.rs index e95f57dbd..761fe9331 100644 --- a/dash-spv/src/sync/headers/validation.rs +++ b/dash-spv/src/sync/headers/validation.rs @@ -4,24 +4,24 @@ use rayon::prelude::*; use std::time::Instant; use crate::error::{ValidationError, ValidationResult}; -use crate::types::CachedHeader; +use crate::types::HashedBlockHeader; /// Validate a chain of headers. -pub fn validate_headers(headers: &[CachedHeader]) -> ValidationResult<()> { +pub fn validate_headers(hashed_headers: &[HashedBlockHeader]) -> ValidationResult<()> { let start = Instant::now(); // Check PoW of i and continuity of i-1 to i in parallel - headers.par_iter().enumerate().try_for_each(|(i, header)| { + hashed_headers.par_iter().enumerate().try_for_each(|(i, header)| { // For the first header, skip chain continuity check since we don't have i-1 here - if i > 0 && header.prev_blockhash != headers[i - 1].block_hash() { + if i > 0 && header.header().prev_blockhash != *hashed_headers[i - 1].hash() { return Err(ValidationError::InvalidHeaderChain(format!( "Header {:?} does not connect to {:?}", - headers[i - 1], + hashed_headers[i - 1], header ))); } // Check if PoW target is met - if !header.target().is_met_by(header.block_hash()) { + if !header.header().target().is_met_by(*header.hash()) { return Err(ValidationError::InvalidProofOfWork); } Ok(()) @@ -29,7 +29,7 @@ pub fn validate_headers(headers: &[CachedHeader]) -> ValidationResult<()> { tracing::trace!( "Header chain validation passed for {} headers, duration: {:?}", - headers.len(), + hashed_headers.len(), start.elapsed(), ); @@ -40,7 +40,7 @@ pub fn validate_headers(headers: &[CachedHeader]) -> ValidationResult<()> { mod tests { use super::validate_headers; use crate::error::ValidationError; - use crate::types::CachedHeader; + use crate::types::HashedBlockHeader; use dashcore::{ block::{Header as BlockHeader, Version}, blockdata::constants::genesis_block, @@ -51,8 +51,8 @@ mod tests { // Very easy target to pass PoW checks for continuity tests const MAX_TARGET: u32 = 0x2100ffff; - fn create_test_header(prev_hash: dashcore::BlockHash, nonce: u32) -> CachedHeader { - CachedHeader::new(BlockHeader { + fn create_test_header(prev_hash: dashcore::BlockHash, nonce: u32) -> HashedBlockHeader { + HashedBlockHeader::from(BlockHeader { version: Version::from_consensus(1), prev_blockhash: prev_hash, merkle_root: dashcore::TxMerkleNode::all_zeros(), @@ -80,7 +80,7 @@ mod tests { for i in 0..10 { let header = create_test_header(prev_hash, i); - prev_hash = header.block_hash(); + prev_hash = *header.hash(); headers.push(header); } @@ -90,7 +90,7 @@ mod tests { #[test] fn test_broken_chain() { let header1 = create_test_header(dashcore::BlockHash::all_zeros(), 0); - let header2 = create_test_header(header1.block_hash(), 1); + let header2 = create_test_header(*header1.hash(), 1); // header3 doesn't connect to header2 let header3 = create_test_header(dashcore::BlockHash::all_zeros(), 2); @@ -100,7 +100,7 @@ mod tests { #[test] fn test_invalid_pow() { - let header = CachedHeader::new(BlockHeader { + let header = HashedBlockHeader::from(BlockHeader { version: Version::from_consensus(1), prev_blockhash: dashcore::BlockHash::all_zeros(), merkle_root: dashcore::TxMerkleNode::all_zeros(), @@ -116,7 +116,7 @@ mod tests { #[test] fn test_genesis_blocks() { for network in [Network::Dash, Network::Testnet, Network::Regtest] { - let genesis = CachedHeader::new(genesis_block(network).header); + let genesis = HashedBlockHeader::from(genesis_block(network).header); assert!( validate_headers(&[genesis]).is_ok(), "Genesis block for {:?} should validate", @@ -128,19 +128,19 @@ mod tests { #[test] fn test_invalid_pow_mid_chain() { let header1 = create_test_header(dashcore::BlockHash::all_zeros(), 0); - let header2 = create_test_header(header1.block_hash(), 1); + let header2 = create_test_header(*header1.hash(), 1); // Header 3 has valid continuity but impossible PoW target - let header3 = CachedHeader::new(BlockHeader { + let header3 = HashedBlockHeader::from(BlockHeader { version: Version::from_consensus(1), - prev_blockhash: header2.block_hash(), + prev_blockhash: *header2.hash(), merkle_root: dashcore::TxMerkleNode::all_zeros(), time: 0, bits: CompactTarget::from_consensus(0x1d00ffff), // Hard target nonce: 0, }); - let header4 = create_test_header(header3.block_hash(), 3); + let header4 = create_test_header(*header3.hash(), 3); let result = validate_headers(&[header1, header2, header3, header4]); assert!(matches!(result, Err(ValidationError::InvalidProofOfWork))); diff --git a/dash-spv/src/types.rs b/dash-spv/src/types.rs index 47dff94cd..c4ff0803f 100644 --- a/dash-spv/src/types.rs +++ b/dash-spv/src/types.rs @@ -18,12 +18,14 @@ use std::time::{Duration, Instant, SystemTime}; use dashcore::{ - block::Header as BlockHeader, hash_types::FilterHeader, network::constants::NetworkExt, - sml::masternode_list_engine::MasternodeListEngine, Amount, BlockHash, Network, Transaction, - Txid, + block::Header as BlockHeader, + consensus::{Decodable, Encodable}, + hash_types::FilterHeader, + network::constants::NetworkExt, + sml::masternode_list_engine::MasternodeListEngine, + Amount, BlockHash, Network, Transaction, Txid, }; use serde::{Deserialize, Serialize}; -use std::sync::Arc; /// Shared, mutex-protected set of filter heights used across components. /// @@ -47,55 +49,65 @@ pub type SharedFilterHeights = std::sync::Arc>, + hash: BlockHash, } -impl CachedHeader { - /// Create a new cached header from a block header - pub fn new(header: BlockHeader) -> Self { - Self { - header, - hash: Arc::new(std::sync::OnceLock::new()), - } - } - - /// Get the block header +impl HashedBlockHeader { pub fn header(&self) -> &BlockHeader { &self.header } - /// Get the cached block hash (computes once, returns cached value thereafter) - pub fn block_hash(&self) -> BlockHash { - *self.hash.get_or_init(|| self.header.block_hash()) + pub fn hash(&self) -> &BlockHash { + &self.hash } +} - /// Convert back to a plain BlockHeader - pub fn into_inner(self) -> BlockHeader { - self.header +impl From for HashedBlockHeader { + fn from(header: BlockHeader) -> Self { + Self { + header, + hash: header.block_hash(), + } } } -impl From for CachedHeader { - fn from(header: BlockHeader) -> Self { - Self::new(header) +impl From<&BlockHeader> for HashedBlockHeader { + fn from(header: &BlockHeader) -> Self { + Self { + header: *header, + hash: header.block_hash(), + } } } -impl AsRef for CachedHeader { - fn as_ref(&self) -> &BlockHeader { - &self.header +impl PartialEq for HashedBlockHeader { + fn eq(&self, other: &Self) -> bool { + self.header == other.header } } -impl std::ops::Deref for CachedHeader { - type Target = BlockHeader; +impl Encodable for HashedBlockHeader { + #[inline] + fn consensus_encode( + &self, + writer: &mut W, + ) -> Result { + Ok(self.header().consensus_encode(writer)? + self.hash().consensus_encode(writer)?) + } +} - fn deref(&self) -> &Self::Target { - &self.header +impl Decodable for HashedBlockHeader { + #[inline] + fn consensus_decode( + reader: &mut R, + ) -> Result { + Ok(Self { + header: BlockHeader::consensus_decode(reader)?, + hash: BlockHash::consensus_decode(reader)?, + }) } }