diff --git a/dash-spv/src/client/lifecycle.rs b/dash-spv/src/client/lifecycle.rs index 4a2215686..4619db179 100644 --- a/dash-spv/src/client/lifecycle.rs +++ b/dash-spv/src/client/lifecycle.rs @@ -71,8 +71,14 @@ impl DashSpvClient Vec::new(), }; let checkpoint_manager = Arc::new(CheckpointManager::new(checkpoints)); - managers.block_headers = - Some(BlockHeadersManager::new(storage.block_headers(), checkpoint_manager).await?); + managers.block_headers = Some( + BlockHeadersManager::new( + storage.block_headers(), + storage.metadata(), + checkpoint_manager, + ) + .await?, + ); if config.enable_filters { managers.filter_headers = Some( diff --git a/dash-spv/src/storage/metadata.rs b/dash-spv/src/storage/metadata.rs index 2c6861ae7..e534dbaf1 100644 --- a/dash-spv/src/storage/metadata.rs +++ b/dash-spv/src/storage/metadata.rs @@ -1,17 +1,25 @@ use std::path::PathBuf; -use async_trait::async_trait; - use crate::{ error::StorageResult, storage::{io::atomic_write, PersistentStorage}, + StorageError, }; +use async_trait::async_trait; +use dashcore::prelude::CoreBlockHeight; + +/// Metadata key for persisting the best known peer height. +const LAST_TARGET_HEIGHT_KEY: &str = "last_target_height"; #[async_trait] pub trait MetadataStorage: Send + Sync + 'static { async fn store_metadata(&mut self, key: &str, value: &[u8]) -> StorageResult<()>; async fn load_metadata(&self, key: &str) -> StorageResult>>; + /// Persist the last target height to metadata storage. + async fn store_last_target_height(&mut self, height: CoreBlockHeight) -> StorageResult<()>; + /// Load the last target height from metadata storage. + async fn load_last_target_height(&self) -> StorageResult; } pub struct PersistentMetadataStorage { @@ -59,4 +67,44 @@ impl MetadataStorage for PersistentMetadataStorage { let data = tokio::fs::read(path).await?; Ok(Some(data)) } + + /// Persist the last target height to metadata storage. + async fn store_last_target_height(&mut self, height: CoreBlockHeight) -> StorageResult<()> { + match serde_json::to_vec(&height) { + Ok(converted) => self.store_metadata(LAST_TARGET_HEIGHT_KEY, &converted).await, + Err(e) => { + let error = format!("Failed to serialize last target height: {}", e); + tracing::warn!(error); + Err(StorageError::Serialization(error)) + } + } + } + + /// Load the last target height from metadata storage. Used by the block headers manager to + /// restore progress after restart. + async fn load_last_target_height(&self) -> StorageResult { + match self.load_metadata(LAST_TARGET_HEIGHT_KEY).await { + Ok(Some(bytes)) => match serde_json::from_slice::(&bytes) { + Ok(last_target_height) => { + tracing::debug!("Restored last target height {}", last_target_height); + Ok(last_target_height) + } + Err(e) => { + let error = format!("Failed to deserialize last target height: {}", e); + tracing::warn!(error); + Err(StorageError::Serialization(error)) + } + }, + Ok(None) => { + let error = "No last target height found (fresh start)".to_string(); + tracing::debug!(error); + Err(StorageError::NotFound(error)) + } + Err(e) => { + let error = format!("Failed to load last target height: {}", e); + tracing::warn!(error); + Err(StorageError::Corruption(error)) + } + } + } } diff --git a/dash-spv/src/storage/mod.rs b/dash-spv/src/storage/mod.rs index 1efb6c996..81d3044e9 100644 --- a/dash-spv/src/storage/mod.rs +++ b/dash-spv/src/storage/mod.rs @@ -14,8 +14,14 @@ mod peers; mod segments; mod transactions; +use crate::error::StorageResult; +use crate::storage::lockfile::LockFile; +use crate::storage::transactions::PersistentTransactionStorage; +use crate::types::{HashedBlock, HashedBlockHeader, MempoolState, UnconfirmedTransaction}; +use crate::ClientConfig; use async_trait::async_trait; use dashcore::hash_types::FilterHeader; +use dashcore::prelude::CoreBlockHeight; use dashcore::{Header as BlockHeader, Txid}; use std::collections::HashMap; use std::ops::Range; @@ -24,12 +30,6 @@ use std::sync::Arc; use std::time::Duration; use tokio::sync::RwLock; -use crate::error::StorageResult; -use crate::storage::lockfile::LockFile; -use crate::storage::transactions::PersistentTransactionStorage; -use crate::types::{HashedBlock, HashedBlockHeader, MempoolState, UnconfirmedTransaction}; -use crate::ClientConfig; - pub use crate::storage::block_headers::{ BlockHeaderStorage, BlockHeaderTip, PersistentBlockHeaderStorage, }; @@ -434,6 +434,14 @@ impl metadata::MetadataStorage for DiskStorageManager { async fn load_metadata(&self, key: &str) -> StorageResult>> { self.metadata.read().await.load_metadata(key).await } + + async fn store_last_target_height(&mut self, height: CoreBlockHeight) -> StorageResult<()> { + self.metadata.write().await.store_last_target_height(height).await + } + + async fn load_last_target_height(&self) -> StorageResult { + self.metadata.read().await.load_last_target_height().await + } } #[async_trait] diff --git a/dash-spv/src/sync/block_headers/manager.rs b/dash-spv/src/sync/block_headers/manager.rs index cd2c30ef6..592e07f67 100644 --- a/dash-spv/src/sync/block_headers/manager.rs +++ b/dash-spv/src/sync/block_headers/manager.rs @@ -13,7 +13,7 @@ use std::time::Instant; use crate::chain::CheckpointManager; use crate::error::{SyncError, SyncResult}; use crate::network::RequestSender; -use crate::storage::{BlockHeaderStorage, BlockHeaderTip}; +use crate::storage::{BlockHeaderStorage, BlockHeaderTip, MetadataStorage}; use crate::sync::block_headers::HeadersPipeline; use crate::sync::{BlockHeadersProgress, ProgressPercentage, SyncEvent, SyncManager, SyncState}; use crate::types::HashedBlockHeader; @@ -30,18 +30,20 @@ use tokio::sync::RwLock; /// - Post-sync header updates via inventory announcements /// /// Generic over `H: BlockHeaderStorage` to allow different storage implementations. -pub struct BlockHeadersManager { +pub struct BlockHeadersManager { /// Current progress of the manager. pub(super) progress: BlockHeadersProgress, /// Block header storage. pub(super) header_storage: Arc>, + /// Metadata storage for persisting the best peer tip height. + pub(super) metadata_storage: Arc>, /// Pipeline for parallel header downloads (used for both initial sync and post-sync). pub(super) pipeline: HeadersPipeline, /// Pending block announcements waiting for headers message (post-sync). pub(super) pending_announcements: HashMap, } -impl std::fmt::Debug for BlockHeadersManager { +impl std::fmt::Debug for BlockHeadersManager { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { f.debug_struct("BlockHeadersManager") .field("progress", &self.progress) @@ -50,10 +52,11 @@ impl std::fmt::Debug for BlockHeadersManager { } } -impl BlockHeadersManager { +impl BlockHeadersManager { /// Create a new headers manager with the given storage and checkpoint manager. pub async fn new( header_storage: Arc>, + metadata_storage: Arc>, checkpoint_manager: Arc, ) -> SyncResult { let tip = header_storage @@ -63,16 +66,21 @@ impl BlockHeadersManager { .await .ok_or_else(|| SyncError::MissingDependency("No tip in storage".to_string()))?; + // Restore persisted target height, fall back to tip height + let target_height = + metadata_storage.read().await.load_last_target_height().await.unwrap_or(tip.height()); + let mut initial_progress = BlockHeadersProgress::default(); initial_progress.set_state(SyncState::WaitingForConnections); initial_progress.update_tip_height(tip.height()); - initial_progress.update_target_height(tip.height()); + initial_progress.update_target_height(target_height); tracing::info!("BlockHeadersManager initialized at height {}", tip.height()); Ok(Self { progress: initial_progress, header_storage, + metadata_storage, pipeline: HeadersPipeline::new(checkpoint_manager), pending_announcements: HashMap::new(), }) @@ -234,10 +242,13 @@ mod tests { use super::*; use crate::chain::checkpoints::testnet_checkpoints; use crate::network::MessageType; - use crate::storage::{DiskStorageManager, PersistentBlockHeaderStorage, StorageManager}; + use crate::storage::{ + DiskStorageManager, PersistentBlockHeaderStorage, PersistentMetadataStorage, StorageManager, + }; use crate::sync::{ManagerIdentifier, SyncManagerProgress}; - type TestBlockHeadersManager = BlockHeadersManager; + type TestBlockHeadersManager = + BlockHeadersManager; fn create_test_checkpoint_manager() -> Arc { Arc::new(CheckpointManager::new(testnet_checkpoints())) @@ -249,7 +260,7 @@ mod tests { let genesis = Header::dummy_batch(0..1); storage.store_headers(&genesis).await.unwrap(); let checkpoint_manager = create_test_checkpoint_manager(); - BlockHeadersManager::new(storage.block_headers(), checkpoint_manager) + BlockHeadersManager::new(storage.block_headers(), storage.metadata(), checkpoint_manager) .await .expect("Failed to create BlockHeadersManager") } diff --git a/dash-spv/src/sync/block_headers/sync_manager.rs b/dash-spv/src/sync/block_headers/sync_manager.rs index 5f6d60b78..a040c9b1f 100644 --- a/dash-spv/src/sync/block_headers/sync_manager.rs +++ b/dash-spv/src/sync/block_headers/sync_manager.rs @@ -1,6 +1,6 @@ use crate::error::SyncResult; use crate::network::{Message, MessageType, NetworkEvent, RequestSender}; -use crate::storage::BlockHeaderStorage; +use crate::storage::{BlockHeaderStorage, MetadataStorage}; use crate::sync::sync_manager::ensure_not_started; use crate::sync::{ BlockHeadersManager, ManagerIdentifier, ProgressPercentage, SyncEvent, SyncManager, @@ -15,7 +15,7 @@ use std::time::{Duration, Instant}; pub(super) const UNSOLICITED_HEADERS_WAIT_TIMEOUT: Duration = Duration::from_secs(3); #[async_trait] -impl SyncManager for BlockHeadersManager { +impl SyncManager for BlockHeadersManager { fn identifier(&self) -> ManagerIdentifier { ManagerIdentifier::BlockHeader } @@ -152,6 +152,8 @@ impl SyncManager for BlockHeadersManager { { if let Some(best_height) = best_height { self.progress.update_target_height(*best_height); + let mut metadata_storage = self.metadata_storage.write().await; + metadata_storage.store_last_target_height(*best_height).await?; } if *connected_count == 0 { self.stop_sync(); diff --git a/dash-spv/src/sync/sync_coordinator.rs b/dash-spv/src/sync/sync_coordinator.rs index 63d26e94f..17a8ef9d3 100644 --- a/dash-spv/src/sync/sync_coordinator.rs +++ b/dash-spv/src/sync/sync_coordinator.rs @@ -71,7 +71,7 @@ where M: MetadataStorage, W: WalletInterface + 'static, { - pub block_headers: Option>, + pub block_headers: Option>, pub filter_headers: Option>, pub filters: Option>, pub blocks: Option>,