diff --git a/dash-spv/src/sync/filter_headers/manager.rs b/dash-spv/src/sync/filter_headers/manager.rs index eb4f3600f..d2f1e5c90 100644 --- a/dash-spv/src/sync/filter_headers/manager.rs +++ b/dash-spv/src/sync/filter_headers/manager.rs @@ -43,6 +43,17 @@ pub struct FilterHeadersManager } impl FilterHeadersManager { + /// The tip height block-header storage holds right now. + /// + /// Distinct from `progress.block_header_tip_height()`, which is only ever + /// as fresh as the last `BlockHeadersStored` / `BlockHeaderSyncComplete` + /// this manager was handed. Storage can move without either arriving — + /// an out-of-order segment completing promotes a run of buffered headers — + /// so the tick reads storage directly to notice. + pub(super) async fn stored_block_header_tip(&self) -> Option { + self.header_storage.read().await.get_tip_height().await + } + /// Transition to `Synced` and return `FilterHeadersSyncComplete` if block headers /// are done and filter headers have reached the target. Returns `None` if already /// `Synced` or conditions are not met. @@ -249,12 +260,16 @@ impl std::fmt::Debug #[cfg(test)] mod tests { use super::*; - use crate::network::MessageType; + use crate::network::{MessageType, NetworkRequest}; use crate::storage::{ DiskStorageManager, PersistentBlockHeaderStorage, PersistentFilterHeaderStorage, StorageManager, }; use crate::sync::{ManagerIdentifier, SyncManagerProgress}; + use crate::types::HashedBlockHeader; + use dashcore::network::message::NetworkMessage; + use dashcore::{block::Version, BlockHash, CompactTarget, Header as BlockHeader}; + use dashcore_hashes::Hash; type TestFilterHeadersManager = FilterHeadersManager; @@ -379,6 +394,96 @@ mod tests { assert!(!events.iter().any(|e| matches!(e, SyncEvent::FilterHeadersSyncComplete { .. }))); } + /// The tick must notice block-header storage advancing on its own. + /// + /// `handle_new_headers` — the only path that extends the CFHeaders queue — + /// runs off `BlockHeadersStored` / `BlockHeaderSyncComplete`. Storage can + /// move without either arriving, because a segment completing out of order + /// promotes a run of buffered headers. When that happened on a mainnet + /// restore the queue stayed at its old target and filter headers stopped + /// permanently, while block headers and ChainLocks carried on. + #[tokio::test] + async fn test_tick_extends_when_storage_tip_advanced_without_an_event() { + let storage = DiskStorageManager::with_temp_dir().await.unwrap(); + let header_storage = storage.block_headers(); + let mut manager = + FilterHeadersManager::new(header_storage.clone(), storage.filter_headers()) + .await + .expect("Failed to create FilterHeadersManager"); + let (sender, mut rx) = create_test_request_sender(); + + // Mid-sync: this manager was last told the tip was 1000. + manager.progress.update_current_height(1000); + manager.progress.update_target_height(1000); + manager.progress.update_block_header_tip_height(1000); + manager.set_state(SyncState::Syncing); + + // Storage moves past that on its own — no event is delivered. + let mut headers = Vec::new(); + let mut prev = BlockHash::from_byte_array([0u8; 32]); + for nonce in 0..1200u32 { + let header = HashedBlockHeader::from(BlockHeader { + version: Version::from_consensus(1), + prev_blockhash: prev, + merkle_root: dashcore::TxMerkleNode::all_zeros(), + time: 0, + bits: CompactTarget::from_consensus(0x2100ffff), + nonce, + }); + prev = *header.hash(); + headers.push(header); + } + header_storage.write().await.store_headers_at_height(&headers, 0).await.unwrap(); + let stored_tip = manager.stored_block_header_tip().await.expect("storage has a tip"); + assert!(stored_tip > 1000, "test setup: storage tip must exceed the known tip"); + + let manager_ref: &mut TestSyncManager = &mut manager; + manager_ref.tick(&sender).await.unwrap(); + + assert_eq!( + manager.progress.block_header_tip_height(), + stored_tip, + "tick must pick up a tip that advanced without an event" + ); + + // The tip alone is not the fix. `handle_new_headers` updates that + // field before it touches the pipeline, so a version that noticed the + // advance and then failed to queue anything would still satisfy the + // assertion above — and would leave sync exactly as stuck as before. + // What has to be true is that a request for the new range went out. + let mut requested_stops = Vec::new(); + while let Ok(request) = rx.try_recv() { + match request { + NetworkRequest::SendMessage(NetworkMessage::GetCFHeaders(get)) => { + requested_stops.push(get.stop_hash); + } + NetworkRequest::SendMessageToPeer(NetworkMessage::GetCFHeaders(get), _) => { + requested_stops.push(get.stop_hash); + } + _ => {} + } + } + assert!( + !requested_stops.is_empty(), + "tick must queue and send CFHeaders requests for the newly available range" + ); + // Every stop hash must be a header the new range actually contains, so + // a request built from the stale target cannot pass this. + let storage = header_storage.read().await; + for stop in &requested_stops { + let mut found = false; + for height in 1001..=stored_tip { + if let Ok(Some(header)) = storage.get_header(height).await { + if header.hash() == stop { + found = true; + break; + } + } + } + assert!(found, "CFHeaders stop hash {stop} is not in the newly discovered range"); + } + } + #[tokio::test] async fn test_on_disconnect() { let mut manager = create_test_manager().await; diff --git a/dash-spv/src/sync/filter_headers/sync_manager.rs b/dash-spv/src/sync/filter_headers/sync_manager.rs index d33d70864..353bf2b8a 100644 --- a/dash-spv/src/sync/filter_headers/sync_manager.rs +++ b/dash-spv/src/sync/filter_headers/sync_manager.rs @@ -156,6 +156,30 @@ impl SyncManager for FilterHeade // Send pending requests (including retries) self.pipeline.send_pending(requests)?; + // Re-read the block-header tip and extend from here if it moved. + // + // `handle_new_headers` — the only thing that ever extends the + // CFHeaders queue — runs solely off `BlockHeaderSyncComplete` and + // `BlockHeadersStored`. Header storage can advance without either + // reaching this manager: a segment that completes out of order + // promotes a run of buffered headers, and on a mainnet scan that is + // how the last stretch of the chain lands. When it does, this + // manager keeps the target it was last told about, its queue drains, + // and nothing re-arms it — filter headers stop for good while block + // headers, ChainLocks and inv announcements carry on, so the client + // looks alive while sync is frozen. Observed on a mainnet restore: + // the queue was last extended to height 2_398_000, block headers + // reached 2_523_515 twenty-four minutes later, and filter headers + // never moved again. + // + // Same fix as promoting finished header segments from the tick + // (#960): trust the tick, not the message. + if let Some(tip) = self.stored_block_header_tip().await { + if tip > self.progress.block_header_tip_height() { + return self.handle_new_headers(tip, requests).await; + } + } + Ok(vec![]) }