fix(dash-spv): preserve buffered block headers across disconnect (#702)

`BlockHeadersManager::clear_in_flight_state` rebuilt the entire `HeadersPipeline` on every disconnect, throwing away each segment's `current_tip_hash`, `current_height`, `buffered_headers`, and `complete` flags.

This PR splits it so that `SegmentState::clear_in_flight` and `HeadersPipeline::clear_in_flight` now reset only the `DownloadCoordinator` per segment. `BlockHeadersManager::clear_in_flight_state` calls the pipeline-level helper instead of constructing a new pipeline. `start_sync` skips `pipeline.init` when the pipeline is already initialized and calls `reset_tip_segment` so a previously synced tip is prepared to let `send_pending` re-fire `GetHeaders`.

Adds unit coverage for the new clear path at the segment, pipeline, and manager levels (mid-sync and post-sync disconnect/reconnect cycles), plus a header-progress monotonicity assertion in `run_disconnect_loop` so the dashd integration tests catch a future regression that drops validated chain state on disconnect.

Author: xdustinface

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

@@ -263,6 +263,7 @@ mod tests {
         DiskStorageManager, PersistentBlockHeaderStorage, PersistentMetadataStorage, StorageManager,
     };
     use crate::sync::{ManagerIdentifier, SyncManager, SyncManagerProgress};
+    use dashcore::network::message::NetworkMessage;
     use tokio::sync::mpsc::unbounded_channel;
 
     type TestBlockHeadersManager =
@@ -428,6 +429,111 @@ mod tests {
         assert!(rx.try_recv().is_err());
     }
 
+    #[tokio::test]
+    async fn test_disconnect_preserves_pipeline_and_resumes_from_advanced_tip() {
+        let mut manager = create_test_manager().await;
+        let (requests, mut rx) = create_test_request_sender();
+
+        // Use a target below the first testnet checkpoint (50000) so the
+        // pipeline produces a single open-ended tip segment.
+        let initial_event = NetworkEvent::PeersUpdated {
+            connected_count: 1,
+            best_height: Some(40_000),
+            addresses: vec![],
+        };
+        manager.handle_network_event(&initial_event, &requests).await.unwrap();
+        assert_eq!(manager.state(), SyncState::Syncing);
+        assert!(manager.pipeline.is_initialized());
+        assert_eq!(manager.pipeline.segment_count(), 1);
+
+        let initial_locator = match rx.try_recv().expect("initial GetHeaders not sent") {
+            NetworkRequest::SendMessage(NetworkMessage::GetHeaders(msg)) => msg.locator_hashes[0],
+            other => panic!("Expected GetHeaders, got {:?}", other),
+        };
+        assert!(rx.try_recv().is_err());
+
+        // Simulate a peer response. The single tip segment drains its buffer
+        // through take_ready_to_store, advancing the storage tip and the
+        // segment's current_tip_hash to advanced_hash.
+        let header = Header::dummy_chain(1, initial_locator).remove(0);
+        let advanced_hash = header.block_hash();
+        manager.handle_headers_pipeline(&[header], &requests).await.unwrap();
+
+        // Drain the follow-up GetHeaders that send_pending issued.
+        match rx.try_recv().expect("follow-up GetHeaders not sent") {
+            NetworkRequest::SendMessage(NetworkMessage::GetHeaders(msg)) => {
+                assert_eq!(msg.locator_hashes[0], advanced_hash);
+            }
+            other => panic!("Expected GetHeaders, got {:?}", other),
+        }
+        assert!(rx.try_recv().is_err());
+
+        let disconnect_event = NetworkEvent::PeersUpdated {
+            connected_count: 0,
+            best_height: Some(40_000),
+            addresses: vec![],
+        };
+        manager.handle_network_event(&disconnect_event, &requests).await.unwrap();
+        assert_eq!(manager.state(), SyncState::WaitingForConnections);
+        assert!(
+            manager.pipeline.is_initialized(),
+            "pipeline must survive disconnect so resume can reuse validated state"
+        );
+        assert_eq!(manager.pipeline.segment_count(), 1);
+
+        // Reconnect: start_sync must skip pipeline.init and resume by sending
+        // GetHeaders from each segment's preserved current_tip_hash.
+        manager.handle_network_event(&initial_event, &requests).await.unwrap();
+        assert_eq!(manager.state(), SyncState::Syncing);
+
+        let resumed_locator = match rx.try_recv().expect("resumed GetHeaders not sent") {
+            NetworkRequest::SendMessage(NetworkMessage::GetHeaders(msg)) => msg.locator_hashes[0],
+            other => panic!("Expected GetHeaders, got {:?}", other),
+        };
+        assert_eq!(
+            resumed_locator, advanced_hash,
+            "GetHeaders on reconnect must use the preserved current_tip_hash"
+        );
+        assert_ne!(resumed_locator, initial_locator);
+        assert!(rx.try_recv().is_err());
+    }
+
+    #[tokio::test]
+    async fn test_disconnect_after_sync_resumes_and_catches_up() {
+        let mut manager = create_synced_manager().await;
+        let tip = manager.tip().await.unwrap();
+        let synced_hash = *tip.hash();
+        manager.pipeline.mark_tip_complete();
+        assert!(manager.pipeline.is_tip_complete());
+
+        let (requests, mut rx) = create_test_request_sender();
+
+        let disconnect_event = NetworkEvent::PeersUpdated {
+            connected_count: 0,
+            best_height: Some(tip.height()),
+            addresses: vec![],
+        };
+        manager.handle_network_event(&disconnect_event, &requests).await.unwrap();
+        assert_eq!(manager.state(), SyncState::WaitingForConnections);
+        assert!(manager.pipeline.is_initialized());
+
+        // Reconnect with a higher peer best_height (a new block was mined).
+        let reconnect_event = NetworkEvent::PeersUpdated {
+            connected_count: 1,
+            best_height: Some(tip.height() + 1),
+            addresses: vec![],
+        };
+        manager.handle_network_event(&reconnect_event, &requests).await.unwrap();
+        assert_eq!(manager.state(), SyncState::Syncing);
+
+        let resumed_locator = match rx.try_recv().expect("resumed GetHeaders not sent") {
+            NetworkRequest::SendMessage(NetworkMessage::GetHeaders(msg)) => msg.locator_hashes[0],
+            other => panic!("Expected GetHeaders, got {:?}", other),
+        };
+        assert_eq!(resumed_locator, synced_hash);
+        assert!(rx.try_recv().is_err());
+    }
+
     #[tokio::test]
     async fn test_empty_headers_after_tip_announcement_is_harmless() {
         let mut manager = create_synced_manager().await;

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

@@ -52,11 +52,6 @@ impl HeadersPipeline {
         }
     }
 
-    /// Get a reference to the checkpoint manager.
-    pub fn checkpoint_manager(&self) -> &Arc<CheckpointManager> {
-        &self.checkpoint_manager
-    }
-
     /// Initialize the pipeline for downloading from current_height to target_height.
     pub fn init(&mut self, current_height: u32, current_hash: BlockHash, target_height: u32) {
         self.segments.clear();
@@ -273,6 +268,18 @@ impl HeadersPipeline {
         }
     }
 
+    /// Drop only per-peer in-flight bookkeeping across every segment.
+    ///
+    /// Buffered headers, segment topology, and per-segment validated tip state
+    /// are preserved. `next_to_store` and `initialized` stay put so a reconnect
+    /// can resume sending `GetHeaders` from each segment's preserved
+    /// `current_tip_hash` without re-fetching what we already have.
+    pub fn clear_in_flight(&mut self) {
+        for segment in &mut self.segments {
+            segment.clear_in_flight();
+        }
+    }
+
     /// Check if pipeline is initialized.
     pub fn is_initialized(&self) -> bool {
         self.initialized
@@ -528,6 +535,68 @@ mod tests {
         assert!(pipeline.segments[0].buffered_headers.is_empty());
     }
 
+    #[test]
+    fn test_clear_in_flight_preserves_buffers_across_segments() {
+        let shared_hash = BlockHash::dummy(42);
+
+        let mut completed =
+            SegmentState::new(0, 0, BlockHash::dummy(0), Some(100), Some(shared_hash));
+        completed.complete = true;
+        completed.current_height = 100;
+        completed.current_tip_hash = shared_hash;
+        // Buffered headers on a complete-but-not-yet-drained segment must survive.
+        let mut completed_header = Header::dummy(1);
+        completed_header.prev_blockhash = BlockHash::dummy(0);
+        completed.buffered_headers.push(HashedBlockHeader::from(completed_header));
+
+        let mut mid = SegmentState::new(1, 100, shared_hash, Some(200), None);
+        mid.coordinator.mark_sent(&[shared_hash]);
+        let mut mid_header = Header::dummy(2);
+        mid_header.prev_blockhash = shared_hash;
+        mid.receive_headers(&[mid_header]).unwrap();
+        let mid_preserved_tip = mid.current_tip_hash;
+        let mid_preserved_height = mid.current_height;
+        let mid_preserved_buffered = mid.buffered_headers.len();
+        // Simulate a fresh in-flight follow-up request for this segment.
+        mid.coordinator.mark_sent(&[mid_preserved_tip]);
+
+        let tip_hash = BlockHash::dummy(99);
+        let mut tip = SegmentState::new(2, 500, tip_hash, None, None);
+        tip.coordinator.mark_sent(&[tip_hash]);
+
+        let cm = create_test_checkpoint_manager(true);
+        let mut pipeline = HeadersPipeline::new(cm);
+        pipeline.initialized = true;
+        pipeline.next_to_store = 0;
+        pipeline.segments = vec![completed, mid, tip];
+
+        pipeline.clear_in_flight();
+
+        // initialized and next_to_store stay put.
+        assert!(pipeline.is_initialized());
+        assert_eq!(pipeline.next_to_store, 0);
+
+        // Completed segment keeps its buffer and complete flag.
+        assert!(pipeline.segments[0].complete);
+        assert_eq!(pipeline.segments[0].buffered_headers.len(), 1);
+        assert_eq!(pipeline.segments[0].current_tip_hash, shared_hash);
+
+        // Mid-download segment: validated chain state preserved; coordinator wiped.
+        assert_eq!(pipeline.segments[1].current_tip_hash, mid_preserved_tip);
+        assert_eq!(pipeline.segments[1].current_height, mid_preserved_height);
+        assert_eq!(pipeline.segments[1].buffered_headers.len(), mid_preserved_buffered);
+        assert!(!pipeline.segments[1].complete);
+        assert_eq!(pipeline.segments[1].coordinator.active_count(), 0);
+        assert_eq!(pipeline.segments[1].coordinator.pending_count(), 0);
+        // can_send returns true so a fresh GetHeaders can resume from preserved tip.
+        assert!(pipeline.segments[1].can_send());
+
+        // Tip segment: in-flight cleared, preserved hash/height intact.
+        assert_eq!(pipeline.segments[2].current_tip_hash, tip_hash);
+        assert_eq!(pipeline.segments[2].coordinator.active_count(), 0);
+        assert!(pipeline.segments[2].can_send());
+    }
+
     #[test]
     fn test_tip_complete_lifecycle() {
         let cm = create_test_checkpoint_manager(true);

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

@@ -194,6 +194,15 @@ impl SegmentState {
             self.coordinator.enqueue_retry(hash);
         }
     }
+
+    /// Drop only per-peer in-flight bookkeeping.
+    ///
+    /// Buffered headers and the validated `current_tip_hash` / `current_height`
+    /// are preserved so a reconnect can resume from where the last peer left off
+    /// without re-fetching headers we already have.
+    pub(super) fn clear_in_flight(&mut self) {
+        self.coordinator.clear();
+    }
 }
 
 #[cfg(test)]
@@ -366,4 +375,41 @@ mod tests {
         }
         assert!(segment.buffered_headers.is_empty());
     }
+
+    #[test]
+    fn test_clear_in_flight_preserves_chain_state() {
+        let start_hash = BlockHash::dummy(0);
+        let mut segment = SegmentState::new(0, 0, start_hash, None, None);
+        segment.coordinator.mark_sent(&[start_hash]);
+
+        let mut header = Header::dummy(1);
+        header.prev_blockhash = start_hash;
+        segment.receive_headers(&[header]).unwrap();
+
+        let preserved_tip_hash = segment.current_tip_hash;
+        let preserved_height = segment.current_height;
+        let preserved_buffered = segment.buffered_headers.len();
+        assert_ne!(preserved_tip_hash, start_hash);
+        assert_eq!(preserved_height, 1);
+        assert_eq!(preserved_buffered, 1);
+
+        // Simulate a fresh in-flight request, then clear it.
+        segment.coordinator.mark_sent(&[preserved_tip_hash]);
+        assert!(segment.coordinator.is_in_flight(&preserved_tip_hash));
+
+        segment.clear_in_flight();
+
+        assert!(!segment.coordinator.is_in_flight(&preserved_tip_hash));
+        assert_eq!(segment.coordinator.active_count(), 0);
+        assert_eq!(segment.coordinator.pending_count(), 0);
+
+        assert_eq!(segment.current_tip_hash, preserved_tip_hash);
+        assert_eq!(segment.current_height, preserved_height);
+        assert_eq!(segment.buffered_headers.len(), preserved_buffered);
+        assert!(!segment.complete);
+
+        // After clearing, can_send should be true again so a fresh GetHeaders
+        // can resume from the preserved tip hash without re-fetching what we have.
+        assert!(segment.can_send());
+    }
 }

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

@@ -1,7 +1,6 @@
 use crate::error::SyncResult;
 use crate::network::{Message, MessageType, NetworkEvent, RequestSender};
 use crate::storage::{BlockHeaderStorage, MetadataStorage};
-use crate::sync::block_headers::pipeline::HeadersPipeline;
 use crate::sync::sync_manager::ensure_not_started;
 use crate::sync::{
     BlockHeadersManager, ManagerIdentifier, ProgressPercentage, SyncEvent, SyncManager,
@@ -38,8 +37,12 @@ impl<H: BlockHeaderStorage, M: MetadataStorage> SyncManager for BlockHeadersMana
     }
 
     fn clear_in_flight_state(&mut self) {
-        let checkpoint_manager = self.pipeline.checkpoint_manager().clone();
-        self.pipeline = HeadersPipeline::new(checkpoint_manager);
+        // Drop only per-peer in-flight bookkeeping. Segment topology and
+        // validated chain state per segment (current_tip_hash, current_height,
+        // buffered_headers, complete) are preserved so a reconnect can resume
+        // from where the disconnected peer left off without re-fetching headers
+        // we already have.
+        self.pipeline.clear_in_flight();
         self.pending_announcements.clear();
         self.announced_peers.clear();
     }
@@ -48,18 +51,31 @@ impl<H: BlockHeaderStorage, M: MetadataStorage> SyncManager for BlockHeadersMana
         ensure_not_started(self.state(), self.identifier())?;
         self.progress.set_state(SyncState::Syncing);
 
-        let tip = self.tip().await?;
-        let target_height = self.progress.target_height();
-
-        // Initialize the pipeline with checkpoint-based segments
-        self.pipeline.init(tip.height(), *tip.hash(), target_height);
-
-        tracing::info!(
-            "Starting parallel header sync from {} to {} ({} segments)",
-            tip.height(),
-            target_height,
-            self.pipeline.segment_count()
-        );
+        if !self.pipeline.is_initialized() {
+            let tip = self.tip().await?;
+            let target_height = self.progress.target_height();
+
+            // Initialize the pipeline with checkpoint-based segments
+            self.pipeline.init(tip.height(), *tip.hash(), target_height);
+
+            tracing::info!(
+                "Starting parallel header sync from {} to {} ({} segments)",
+                tip.height(),
+                target_height,
+                self.pipeline.segment_count()
+            );
+        } else {
+            // Resume path: if we previously synced past the tip the open-ended
+            // segment is marked complete and `send_pending` would skip it.
+            // Reset it so a fresh GetHeaders is fired from the preserved
+            // `current_tip_hash`. No-op if the tip is still mid-sync.
+            self.pipeline.reset_tip_segment();
+            tracing::info!(
+                "Resuming parallel header sync ({} segments, {} buffered)",
+                self.pipeline.segment_count(),
+                self.pipeline.total_buffered()
+            );
+        }
 
         // Send initial batch of requests
         let sent = self.pipeline.send_pending(requests)?;

dash-spv/tests/dashd_sync/helpers.rs

@@ -16,6 +16,11 @@ use dash_spv::test_utils::SYNC_TIMEOUT;
 
 use super::setup::{ClientHandle, TestContext};
 
+/// Read the headers manager's effective height (storage tip plus buffered).
+fn current_header_height(handle: &ClientHandle) -> u32 {
+    handle.progress_receiver.borrow().headers().ok().map(|h| h.current_height()).unwrap_or(0)
+}
+
 /// Wait for sync to reach target height.
 pub(super) async fn wait_for_sync(
     progress_receiver: &mut watch::Receiver<SyncProgress>,
@@ -198,7 +203,9 @@ pub(super) async fn assert_no_mempool_tx(
 ///
 /// Waits for progress events, disconnects all peers after every 5th event,
 /// validates disconnect/reconnect network events, and asserts wallet state
-/// after sync completes.
+/// after sync completes. Also asserts header progress (storage tip plus
+/// buffered) is monotonic across each disconnect cycle, so a regression that
+/// drops validated chain state on disconnect is caught.
 pub(super) async fn run_disconnect_loop(
     mut client_handle: ClientHandle,
     node: &DashCoreNode,
@@ -230,6 +237,7 @@ pub(super) async fn run_disconnect_loop(
                                 disconnect_count + 1,
                                 event.description()
                             );
+                            let pre_disconnect_height = current_header_height(&client_handle);
                             node.disconnect_all_peers();
                             disconnect_count += 1;
                             events_since_disconnect = 0;
@@ -249,6 +257,13 @@ pub(super) async fn run_disconnect_loop(
                             ).await;
                             assert!(saw_reconnect, "SPV should reconnect after disconnection");
                             tracing::info!("SPV reconnected (PeerConnected)");
+
+                            let post_reconnect_height = current_header_height(&client_handle);
+                            assert!(
+                                post_reconnect_height >= pre_disconnect_height,
+                                "Header progress regressed across disconnect {}: {} -> {}",
+                                disconnect_count, pre_disconnect_height, post_reconnect_height
+                            );
                         }
                     }
                     Ok(SyncEvent::SyncComplete { .. }) => {