fix(spv,sml): harden sync robustness and masternode QRInfo handling

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

@@ -263,6 +263,43 @@ impl<H: BlockHeaderStorage, M: MetadataStorage> BlockHeadersManager<H, M> {
         }])
     }
 
+    /// Fire one fallback `GetHeaders` for block announcements that have gone
+    /// unanswered past the wait timeout, then drop them.
+    ///
+    /// Announced-but-unobtainable hashes (a peer that vanished, an orphan that
+    /// never reconnected) otherwise pin the manager: `finalize_sync_if_complete`
+    /// refuses to finish while any announcement is outstanding, so it resets the
+    /// tip segment and re-requests every pass. That recovery loop only aged the
+    /// stale entries out on the `Synced` branch of `tick`, which a `Syncing`
+    /// manager never reaches — so during initial sync the loop ran forever,
+    /// `BlockHeaderSyncComplete` never fired, and every downstream manager
+    /// stalled behind it. Sweeping here, from every state, bounds the loop. (#960)
+    pub(super) fn prune_stale_announcements(&mut self, requests: &RequestSender) -> SyncResult<()> {
+        let now = Instant::now();
+        let stale: Vec<BlockHash> = self
+            .pending_announcements
+            .iter()
+            .filter(|(_, announced_at)| {
+                now.duration_since(**announced_at)
+                    > super::sync_manager::UNSOLICITED_HEADERS_WAIT_TIMEOUT
+            })
+            .map(|(hash, _)| *hash)
+            .collect();
+
+        if stale.is_empty() {
+            return Ok(());
+        }
+
+        tracing::info!("Sending fallback GetHeaders for {} stale announcements", stale.len());
+        self.pipeline.reset_tip_segment();
+        self.pipeline.send_pending(requests)?;
+
+        for hash in stale {
+            self.pending_announcements.remove(&hash);
+        }
+        Ok(())
+    }
+
     /// Handle inventory announcements for new blocks.
     ///
     /// During initial sync, Dash Core sends inv (not header announcements) because
@@ -419,6 +456,65 @@ mod tests {
         );
     }
 
+    /// A block announced during initial sync that no peer will ever answer must
+    /// not pin the manager in `Syncing`. `finalize_sync_if_complete` refuses to
+    /// finish while any announcement is outstanding and re-requests it every
+    /// pass; the aging-out sweep that breaks that loop used to run only on the
+    /// `Synced` branch of `tick`, which a `Syncing` manager never reaches, so
+    /// the loop ran forever and `BlockHeaderSyncComplete` never fired. (#960)
+    #[tokio::test]
+    async fn test_syncing_prunes_unobtainable_announcement_and_finalizes() {
+        let mut manager = create_test_manager().await;
+        let tip = manager.tip().await.unwrap();
+        manager.pipeline.init(tip.height(), *tip.hash(), tip.height());
+        manager.progress.set_state(SyncState::Syncing);
+
+        let (requests, mut rx) = create_test_request_sender();
+
+        // Drive the single tip segment complete: nothing left to download.
+        manager.pipeline.send_pending(&requests).unwrap();
+        while rx.try_recv().is_ok() {}
+        manager.pipeline.receive_headers(&[]).unwrap();
+        assert!(manager.pipeline.is_complete());
+
+        // A phantom announcement, already aged past the wait timeout.
+        let phantom = BlockHash::dummy(123);
+        manager.pending_announcements.insert(
+            phantom,
+            std::time::Instant::now()
+                .checked_sub(std::time::Duration::from_secs(30))
+                .expect("monotonic clock has been running longer than 30s"),
+        );
+
+        // First tick: the sweep fires its fallback (which re-opens the tip
+        // segment) and drops the phantom so it can never block completion again.
+        let events = manager.tick(&requests).await.unwrap();
+        assert!(
+            !manager.pending_announcements.contains_key(&phantom),
+            "the stale announcement must be pruned"
+        );
+        assert!(!events.iter().any(|e| matches!(e, SyncEvent::BlockHeaderSyncComplete { .. })));
+        assert_eq!(manager.state(), SyncState::Syncing);
+
+        // The fallback GetHeaders is answered by the peer's tip: an empty
+        // response re-completes the tip segment.
+        while rx.try_recv().is_ok() {}
+        manager.pipeline.receive_headers(&[]).unwrap();
+        assert!(manager.pipeline.is_complete());
+
+        // Second tick: nothing stale remains, so finalize runs and the sync ends.
+        let events = manager.tick(&requests).await.unwrap();
+        assert_eq!(
+            manager.state(),
+            SyncState::Synced,
+            "with the announcement pruned, the tick must finish the sync"
+        );
+        assert!(
+            events.iter().any(|e| matches!(e, SyncEvent::BlockHeaderSyncComplete { .. })),
+            "completion must be announced so downstream managers start"
+        );
+    }
+
     #[tokio::test]
     async fn test_block_headers_manager_new() {
         let manager = create_test_manager().await;

dash-spv/src/sync/block_headers/pipeline.rs

@@ -176,10 +176,19 @@ impl HeadersPipeline {
                     continue;
                 }
                 // If tip segment was completed but receives new headers (post-sync),
-                // reset it so take_ready_to_store() can process the new headers
+                // reset it so take_ready_to_store() can process the new headers.
+                //
+                // Only ever move `next_to_store` back toward the tip, never
+                // forward: `send_pending` requests solely the window
+                // `[next_to_store, next_to_store + ACTIVE_SEGMENT_WINDOW)`, so
+                // jumping it forward past a still-downloading lower segment
+                // drops that segment out of the window for good and header sync
+                // hangs. When every lower segment is already stored
+                // `next_to_store` sits at (or past) this tip, so the min is a
+                // no-op there and only guards the out-of-order case. (#950)
                 if segment.complete && segment.target_height.is_none() {
                     segment.complete = false;
-                    self.next_to_store = idx;
+                    self.next_to_store = self.next_to_store.min(idx);
                     // A headers announcement may contain multiple consecutive headers. The
                     // coordinator tracks the response by its first previous hash, just like a
                     // requested headers batch.
@@ -320,8 +329,13 @@ impl HeadersPipeline {
         for (idx, segment) in self.segments.iter_mut().enumerate() {
             if segment.target_height.is_none() && segment.complete {
                 segment.complete = false;
-                // Reset next_to_store so buffered headers can be processed
-                self.next_to_store = idx;
+                // Reset next_to_store so buffered headers can be processed, but
+                // only ever backward toward the tip. Moving it forward past a
+                // lower segment that has not finished downloading would strand
+                // that segment outside `send_pending`'s active window and hang
+                // header sync; when the lower segments are all stored,
+                // `next_to_store` already sits at or past this tip. (#950)
+                self.next_to_store = self.next_to_store.min(idx);
                 tracing::debug!(
                     "Reset tip segment {} at height {} for continued syncing",
                     segment.segment_id,
@@ -697,6 +711,80 @@ mod tests {
         assert!(pipeline.segments[2].can_send());
     }
 
+    /// `reset_tip_segment` must not jump `next_to_store` forward past a lower
+    /// segment that is still downloading. `send_pending` only requests the
+    /// window `[next_to_store, next_to_store + ACTIVE_SEGMENT_WINDOW)`, so a
+    /// forward jump drops that segment out of the window for good and header
+    /// sync hangs. (#950)
+    #[test]
+    fn test_reset_tip_segment_does_not_skip_incomplete_lower_segment() {
+        let cm = create_test_checkpoint_manager(true);
+        let mut pipeline = HeadersPipeline::new(cm);
+
+        // Segment 0: a checkpoint segment still downloading.
+        let mid =
+            SegmentState::new(0, 0, BlockHash::dummy(0), Some(100), Some(BlockHash::dummy(1)));
+        // Segment 1: the open-ended tip, completed early by an empty response.
+        let mut tip = SegmentState::new(1, 100, BlockHash::dummy(1), None, None);
+        tip.complete = true;
+
+        pipeline.set_segments_for_test(vec![mid, tip]);
+        assert_eq!(pipeline.next_to_store, 0);
+
+        assert!(pipeline.reset_tip_segment(), "the complete tip segment must reset");
+        assert_eq!(
+            pipeline.next_to_store, 0,
+            "reset must not skip the still-downloading lower segment"
+        );
+
+        // The lower segment is still inside the window and gets requested.
+        let (sender, mut rx) = create_test_request_sender();
+        assert!(pipeline.send_pending(&sender).unwrap() >= 1);
+        let mut requested = Vec::new();
+        while let Ok(request) = rx.try_recv() {
+            if let NetworkRequest::SendMessage(
+                dashcore::network::message::NetworkMessage::GetHeaders(get),
+            ) = request
+            {
+                requested.push(get.locator_hashes[0]);
+            }
+        }
+        assert!(
+            requested.contains(&BlockHash::dummy(0)),
+            "the segment the reset skipped over must still be requested"
+        );
+    }
+
+    /// The same guard on the receive-path tip reset: an unsolicited post-sync
+    /// header for the completed tip must not advance `next_to_store` past a
+    /// lower segment that has not finished downloading. (#950)
+    #[test]
+    fn test_receive_tip_reset_does_not_skip_incomplete_lower_segment() {
+        let cm = create_test_checkpoint_manager(true);
+        let mut pipeline = HeadersPipeline::new(cm);
+
+        let tip_hash = BlockHash::dummy(50);
+        let mid =
+            SegmentState::new(0, 0, BlockHash::dummy(0), Some(100), Some(BlockHash::dummy(1)));
+        let mut tip = SegmentState::new(1, 1000, tip_hash, None, None);
+        tip.complete = true;
+
+        pipeline.set_segments_for_test(vec![mid, tip]);
+        assert_eq!(pipeline.next_to_store, 0);
+
+        // A post-sync header extending the tip routes to the tip segment.
+        let mut header = Header::dummy(1);
+        header.prev_blockhash = tip_hash;
+        let matched = pipeline.receive_headers(&[header.into()]).unwrap();
+
+        assert_eq!(matched, Some(1), "the header must route to the tip segment");
+        assert_eq!(
+            pipeline.next_to_store, 0,
+            "the tip reset must not skip the still-downloading lower segment"
+        );
+        assert!(!pipeline.segments[1].complete, "the tip segment must be reset to non-complete");
+    }
+
     #[test]
     fn test_tip_complete_lifecycle() {
         let cm = create_test_checkpoint_manager(true);

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

@@ -8,8 +8,7 @@ use crate::sync::{
 };
 use async_trait::async_trait;
 use dashcore::network::message::NetworkMessage;
-use dashcore::BlockHash;
-use std::time::{Duration, Instant};
+use std::time::Duration;
 
 /// Timeout waiting for unsolicited header messages after a block announcement.
 pub(super) const UNSOLICITED_HEADERS_WAIT_TIMEOUT: Duration = Duration::from_secs(3);
@@ -152,6 +151,12 @@ impl<H: BlockHeaderStorage, M: MetadataStorage> SyncManager for BlockHeadersMana
 
         self.pipeline.handle_timeouts();
 
+        // Age out announcements no peer will answer, in every state. This must
+        // run before the `Syncing` branch's `finalize_sync_if_complete`, which
+        // otherwise loops forever resetting the tip segment for a permanently
+        // unobtainable hash and never reaches `Synced`. (#960)
+        self.prune_stale_announcements(requests)?;
+
         // During initial sync, send more requests and log progress
         if self.state() == SyncState::Syncing {
             // Take, refill, then store — the same order `handle_headers_pipeline`
@@ -177,34 +182,6 @@ impl<H: BlockHeaderStorage, M: MetadataStorage> SyncManager for BlockHeadersMana
             return Ok(events);
         }
 
-        // Post-sync: check for stale block announcements
-        if self.state() == SyncState::Synced {
-            let now = Instant::now();
-            let stale: Vec<BlockHash> = self
-                .pending_announcements
-                .iter()
-                .filter(|(_, announced_at)| {
-                    now.duration_since(**announced_at) > UNSOLICITED_HEADERS_WAIT_TIMEOUT
-                })
-                .map(|(hash, _)| *hash)
-                .collect();
-
-            if !stale.is_empty() {
-                tracing::info!(
-                    "Sending fallback GetHeaders for {} stale announcements",
-                    stale.len()
-                );
-
-                // Reset tip segment and send requests via pipeline
-                self.pipeline.reset_tip_segment();
-                self.pipeline.send_pending(requests)?;
-
-                for hash in stale {
-                    self.pending_announcements.remove(&hash);
-                }
-            }
-        }
-
         Ok(vec![])
     }
 

dash-spv/src/sync/download_coordinator.rs

@@ -152,6 +152,24 @@ impl<K: Hash + Eq + Clone> DownloadCoordinator<K> {
         }
     }
 
+    /// Return items `take_pending` handed out that were never dispatched.
+    ///
+    /// `take_pending` removes items from the queue before the caller sends
+    /// them, and only `mark_sent` moves them into in-flight — so a caller that
+    /// bails out between the two leaves those items in neither tracker, and
+    /// nothing puts them back: `check_timeouts` and `requeue_in_flight` both
+    /// walk in-flight only. This restores them to the front of the queue in
+    /// their original order, so the next send reissues them lowest-first and a
+    /// sequential consumer keeps making progress.
+    ///
+    /// Retry counts are deliberately left alone: the request never reached a
+    /// peer, so no peer failed to answer it.
+    pub(crate) fn return_unsent(&mut self, items: Vec<K>) {
+        for item in items.into_iter().rev() {
+            self.pending.push_front(item);
+        }
+    }
+
     /// Handle a received item.
     ///
     /// Returns true if the item was being tracked, false if unexpected.
@@ -362,6 +380,35 @@ mod tests {
         assert_eq!(requeued, vec![1, 2, 3]);
     }
 
+    /// `return_unsent` must put items `take_pending` handed out but that were
+    /// never dispatched back at the *front* of the queue, in their original
+    /// order, without charging them a retry. Nothing else covers the gap
+    /// between `take_pending` and `mark_sent` — `check_timeouts` and
+    /// `requeue_in_flight` both walk in-flight only — so an item dropped there
+    /// is never requested again. (#972)
+    #[test]
+    fn test_return_unsent_restores_order_without_charging_a_retry() {
+        let mut coord: DownloadCoordinator<u32> = DownloadCoordinator::default();
+        coord.enqueue([1, 2, 3, 4]);
+
+        let taken = coord.take_pending(3);
+        assert_eq!(taken, vec![1, 2, 3]);
+        assert_eq!(coord.pending_count(), 1);
+
+        // The caller dispatched 1, then failed on 2 and gave back the rest.
+        coord.mark_sent(&[1]);
+        coord.return_unsent(taken[1..].to_vec());
+
+        assert!(coord.is_in_flight(&1), "the dispatched item stays in flight");
+        assert_eq!(coord.pending_count(), 3);
+        assert_eq!(
+            coord.take_pending(3),
+            vec![2, 3, 4],
+            "returned items go back ahead of what was never taken, in order"
+        );
+        assert!(coord.retry_counts.is_empty(), "a request that never left is not a retry");
+    }
+
     #[test]
     fn test_requeue_in_flight_preserves_retry_counts() {
         let mut coord: DownloadCoordinator<u32> = DownloadCoordinator::default();