fix(dash-spv): extend the CFHeaders queue from the tick, not only on a header event

dash-spv/src/sync/filter_headers/manager.rs

@@ -43,6 +43,17 @@ pub struct FilterHeadersManager<H: BlockHeaderStorage, FH: FilterHeaderStorage>
 }
 
 impl<H: BlockHeaderStorage, FH: FilterHeaderStorage> FilterHeadersManager<H, FH> {
+    /// 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<u32> {
+        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<H: BlockHeaderStorage, FH: FilterHeaderStorage> 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<PersistentBlockHeaderStorage, PersistentFilterHeaderStorage>;
@@ -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;

dash-spv/src/sync/filter_headers/sync_manager.rs

@@ -156,6 +156,30 @@ impl<H: BlockHeaderStorage, FH: FilterHeaderStorage> 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![])
     }