fix(dash-spv): requeue in-flight downloads when a single peer disconnects

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

@@ -47,6 +47,17 @@ impl<H: BlockHeaderStorage, M: MetadataStorage> SyncManager for BlockHeadersMana
         self.announced_peers.clear();
     }
 
+    fn on_peer_disconnect(&mut self) {
+        // Only the active sync path reissues header requests. Dropping the
+        // in-flight marker in any other state would strand the request rather
+        // than retry it, and would also mask the tip announcement a freshly
+        // connected peer gets while a catch-up request is outstanding.
+        if self.state() != SyncState::Syncing {
+            return;
+        }
+        self.pipeline.clear_in_flight();
+    }
+
     async fn start_sync(&mut self, requests: &RequestSender) -> SyncResult<Vec<SyncEvent>> {
         ensure_not_started(self.state(), self.identifier())?;
         self.progress.set_state(SyncState::Syncing);
@@ -191,6 +202,14 @@ impl<H: BlockHeaderStorage, M: MetadataStorage> SyncManager for BlockHeadersMana
                 address,
             } => {
                 self.announced_peers.remove(address);
+                self.on_peer_disconnect();
+                if self.state() == SyncState::Syncing {
+                    // Reissue before returning to the task loop. A segment
+                    // rejects headers whose request it no longer tracks, so
+                    // leaving the resend to the next tick would open a window
+                    // in which a surviving peer's reply looks unsolicited.
+                    self.pipeline.send_pending(requests)?;
+                }
             }
             NetworkEvent::PeersUpdated {
                 connected_count,

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

@@ -167,16 +167,20 @@ impl<H: BlockHeaderStorage, B: BlockStorage, W: WalletInterface> std::fmt::Debug
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::network::{MessageType, NetworkManager};
+    use crate::network::{MessageType, NetworkEvent, NetworkManager, NetworkRequest};
     use crate::storage::{
         DiskStorageManager, PersistentBlockHeaderStorage, PersistentBlockStorage, StorageManager,
     };
     use crate::sync::{ManagerIdentifier, SyncEvent, SyncManagerProgress};
-    use crate::test_utils::MockNetworkManager;
+    use crate::test_utils::{test_socket_address, MockNetworkManager};
     use crate::types::HashedBlock;
+    use dashcore::network::message::NetworkMessage;
+    use dashcore::network::message_blockdata::Inventory;
+    use dashcore::BlockHash;
     use key_wallet_manager::test_utils::{MockWallet, MOCK_WALLET_ID};
-    use key_wallet_manager::FilterMatchKey;
+    use key_wallet_manager::{FilterMatchKey, WalletId};
     use std::collections::{BTreeMap, BTreeSet};
+    use tokio::sync::mpsc;
 
     type TestBlocksManager =
         BlocksManager<PersistentBlockHeaderStorage, PersistentBlockStorage, MockWallet>;
@@ -234,6 +238,61 @@ mod tests {
         assert!(events.is_empty());
     }
 
+    /// Losing one peer of several must put that peer's `getdata` back on the
+    /// queue right away instead of waiting out the block download timeout.
+    #[tokio::test]
+    async fn test_peer_disconnect_reissues_in_flight_block_requests() {
+        let mut manager = create_test_manager().await;
+        let (tx, mut rx) = mpsc::unbounded_channel();
+        let requests = RequestSender::new(tx);
+
+        let blocks: BTreeMap<FilterMatchKey, BTreeSet<WalletId>> = (100..103)
+            .map(|height| {
+                (
+                    FilterMatchKey::new(height, BlockHash::dummy(height)),
+                    BTreeSet::from([MOCK_WALLET_ID]),
+                )
+            })
+            .collect();
+        let event = SyncEvent::BlocksNeeded {
+            blocks,
+        };
+        manager.handle_sync_event(&event, &requests).await.unwrap();
+
+        let requested = drain_requested_blocks(&mut rx);
+        assert_eq!(requested.len(), 3);
+
+        let disconnect = NetworkEvent::PeerDisconnected {
+            address: test_socket_address(1),
+        };
+        manager.handle_network_event(&disconnect, &requests).await.unwrap();
+
+        manager.tick(&requests).await.unwrap();
+        assert_eq!(drain_requested_blocks(&mut rx), requested);
+    }
+
+    fn drain_requested_blocks(
+        rx: &mut mpsc::UnboundedReceiver<NetworkRequest>,
+    ) -> BTreeSet<BlockHash> {
+        let mut hashes = BTreeSet::new();
+        while let Ok(request) = rx.try_recv() {
+            match request {
+                NetworkRequest::SendMessage(NetworkMessage::GetData(inventory)) => {
+                    for item in inventory {
+                        match item {
+                            Inventory::Block(hash) => {
+                                hashes.insert(hash);
+                            }
+                            other => panic!("Expected a block inventory item, got {:?}", other),
+                        }
+                    }
+                }
+                other => panic!("Expected GetData, got {:?}", other),
+            }
+        }
+        hashes
+    }
+
     /// `process_buffered_blocks` must call `process_block_for_wallets` with
     /// the exact wallet set carried in the pipeline so already-synced
     /// wallets are not touched by routing logic.

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

@@ -62,6 +62,10 @@ impl<H: BlockHeaderStorage, B: BlockStorage, W: WalletInterface + 'static> SyncM
         self.pipeline.requeue_in_flight();
     }
 
+    fn on_peer_disconnect(&mut self) {
+        self.pipeline.requeue_in_flight();
+    }
+
     async fn handle_message(
         &mut self,
         msg: Message,

dash-spv/src/sync/download_coordinator.rs

@@ -375,6 +375,21 @@ mod tests {
         assert_eq!(coord.retry_counts.get(&7), Some(&1));
         assert!(!coord.is_in_flight(&7));
         assert_eq!(coord.pending_count(), 1);
+
+        // A peer disconnect is not an attempt against the item's budget, so
+        // repeated requeues must not spend it.
+        let items = coord.take_pending(1);
+        coord.mark_sent(&items);
+        coord.requeue_in_flight();
+        assert_eq!(coord.retry_counts.get(&7), Some(&1));
+
+        // A genuine timeout still counts, and delivery still clears the budget.
+        let items = coord.take_pending(1);
+        coord.mark_sent(&items);
+        coord.enqueue_retry(7);
+        assert_eq!(coord.retry_counts.get(&7), Some(&2));
+        assert!(coord.receive(&7));
+        assert_eq!(coord.retry_counts.get(&7), None);
     }
 
     #[test]

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

@@ -254,6 +254,16 @@ impl FilterHeadersPipeline {
         ready
     }
 
+    /// Move in-flight `getcfheaders` requests back to pending after a peer
+    /// disconnect so the next `send_pending` reissues them.
+    ///
+    /// `batch_starts` must survive, since `send_pending` errors out on a pending
+    /// stop hash with no start height. `next_expected` and the out-of-order
+    /// buffer survive too, so already-received batches are not re-downloaded.
+    pub(super) fn requeue_in_flight(&mut self) {
+        self.coordinator.requeue_in_flight();
+    }
+
     /// Re-enqueue timed out requests for retry.
     pub(super) fn handle_timeouts(&mut self) {
         for stop_hash in self.coordinator.check_timeouts() {
@@ -267,6 +277,10 @@ mod tests {
     use dashcore_hashes::Hash;
 
     use super::*;
+    use crate::network::NetworkRequest;
+    use dashcore::hash_types::{FilterHash, FilterHeader};
+    use dashcore::network::message_filter::GetCFHeaders;
+    use tokio::sync::mpsc::unbounded_channel;
 
     #[test]
     fn test_cfheaders_pipeline_new() {
@@ -416,6 +430,60 @@ mod tests {
         assert!(matches!(err, SyncError::InvalidState(_)));
     }
 
+    /// A peer disconnect requeues in-flight batches without discarding what the
+    /// pipeline has already made of the ones that came back, so the reissued
+    /// requests carry their original start heights and nothing is re-downloaded.
+    #[test]
+    fn test_requeue_in_flight_reissues_batches_and_keeps_progress() {
+        let mut pipeline = FilterHeadersPipeline::new();
+        pipeline.next_expected = 1;
+        pipeline.target_height = 6000;
+
+        let hash1 = BlockHash::from_byte_array([0x01; 32]);
+        let hash2 = BlockHash::from_byte_array([0x02; 32]);
+        pipeline.coordinator.mark_sent(&[hash1, hash2]);
+        pipeline.batch_starts.insert(hash1, 1);
+        pipeline.batch_starts.insert(hash2, 2001);
+
+        // A third batch already came back out of order and is waiting for the
+        // two above to be processed first.
+        pipeline.buffered.insert(
+            4001,
+            CFHeaders {
+                filter_type: 0,
+                stop_hash: BlockHash::from_byte_array([0x03; 32]),
+                previous_filter_header: FilterHeader::all_zeros(),
+                filter_hashes: vec![FilterHash::all_zeros()],
+            },
+        );
+
+        pipeline.requeue_in_flight();
+        assert_eq!(pipeline.coordinator.active_count(), 0);
+        assert_eq!(pipeline.coordinator.pending_count(), 2);
+
+        let (tx, mut rx) = unbounded_channel();
+        let requests = RequestSender::new(tx);
+        assert_eq!(pipeline.send_pending(&requests).unwrap(), 2);
+
+        let mut reissued = Vec::new();
+        while let Ok(request) = rx.try_recv() {
+            match request {
+                NetworkRequest::SendMessage(NetworkMessage::GetCFHeaders(GetCFHeaders {
+                    start_height,
+                    stop_hash,
+                    ..
+                })) => reissued.push((start_height, stop_hash)),
+                other => panic!("Expected GetCFHeaders, got {:?}", other),
+            }
+        }
+        reissued.sort();
+        assert_eq!(reissued, vec![(1, hash1), (2001, hash2)]);
+
+        assert_eq!(pipeline.next_expected(), 1);
+        assert_eq!(pipeline.buffered.len(), 1);
+        assert_eq!(pipeline.target_height, 6000);
+    }
+
     #[test]
     fn test_handle_timeouts_multiple_batches() {
         use std::time::Duration;

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

@@ -37,6 +37,10 @@ impl<H: BlockHeaderStorage, FH: FilterHeaderStorage> SyncManager for FilterHeade
         self.block_headers_synced = false;
     }
 
+    fn on_peer_disconnect(&mut self) {
+        self.pipeline.requeue_in_flight();
+    }
+
     async fn handle_message(
         &mut self,
         msg: Message,

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

@@ -47,6 +47,10 @@ impl<
         self.filter_pipeline.requeue_in_flight();
     }
 
+    fn on_peer_disconnect(&mut self) {
+        self.filter_pipeline.requeue_in_flight();
+    }
+
     async fn start_sync(&mut self, requests: &RequestSender) -> SyncResult<Vec<SyncEvent>> {
         ensure_not_started(self.state(), self.identifier())?;
 

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

@@ -142,6 +142,15 @@ impl MnListDiffPipeline {
         tracing::debug!("Requeued MnListDiff for {} for retry", diff.block_hash);
     }
 
+    /// Move in-flight `getmnlistd` requests back to pending after a peer
+    /// disconnect so the next `send_pending` reissues them.
+    ///
+    /// `base_hashes` must survive, since `send_pending` drops a pending target
+    /// hash that has no base hash to pair it with.
+    pub(super) fn requeue_in_flight(&mut self) {
+        self.coordinator.requeue_in_flight();
+    }
+
     /// Handle timeouts, re-queuing timed out requests.
     pub(super) fn handle_timeouts(&mut self) {
         for target_hash in self.coordinator.check_timeouts() {
@@ -167,6 +176,10 @@ mod tests {
     use dashcore_hashes::Hash;
 
     use super::*;
+    use crate::network::NetworkRequest;
+    use dashcore::network::message::NetworkMessage;
+    use dashcore::network::message_sml::GetMnListDiff;
+    use tokio::sync::mpsc::unbounded_channel;
 
     /// Create a minimal MnListDiff for testing.
     fn create_test_diff(base_hash: BlockHash, target_hash: BlockHash) -> MnListDiff {
@@ -348,6 +361,48 @@ mod tests {
         assert!(!pipeline.is_complete());
     }
 
+    /// A peer disconnect requeues every in-flight request, and each one must be
+    /// reissued with the base hash it was originally paired with.
+    #[test]
+    fn test_requeue_in_flight_reissues_with_base_hashes() {
+        let mut pipeline = MnListDiffPipeline::new();
+
+        let base1 = BlockHash::from_byte_array([0x01; 32]);
+        let target1 = BlockHash::from_byte_array([0x02; 32]);
+        let base2 = BlockHash::from_byte_array([0x03; 32]);
+        let target2 = BlockHash::from_byte_array([0x04; 32]);
+
+        pipeline.queue_requests(vec![(base1, target1), (base2, target2)]);
+
+        let (tx, mut rx) = unbounded_channel();
+        let requests = RequestSender::new(tx);
+        pipeline.send_pending(&requests).unwrap();
+        assert_eq!(pipeline.active_count(), 2);
+        while rx.try_recv().is_ok() {}
+
+        pipeline.requeue_in_flight();
+        assert_eq!(pipeline.active_count(), 0);
+        assert_eq!(pipeline.coordinator.pending_count(), 2);
+
+        pipeline.send_pending(&requests).unwrap();
+        assert_eq!(pipeline.active_count(), 2);
+
+        let mut reissued = Vec::new();
+        while let Ok(request) = rx.try_recv() {
+            match request {
+                NetworkRequest::SendMessage(NetworkMessage::GetMnListD(GetMnListDiff {
+                    base_block_hash,
+                    block_hash,
+                })) => reissued.push((base_block_hash, block_hash)),
+                other => panic!("Expected GetMnListD, got {:?}", other),
+            }
+        }
+        reissued.sort();
+        let mut expected = vec![(base1, target1), (base2, target2)];
+        expected.sort();
+        assert_eq!(reissued, expected);
+    }
+
     #[test]
     fn test_requeue_always_succeeds() {
         let mut pipeline = MnListDiffPipeline::new();

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

@@ -231,6 +231,12 @@ impl<H: BlockHeaderStorage> SyncManager for MasternodesManager<H> {
         self.sync_state.last_processed_qrinfo_tip = None;
     }
 
+    fn on_peer_disconnect(&mut self) {
+        // The QRInfo request is tracked outside the pipeline with its own
+        // escalating timeout and attempt budget, so it is left to that path.
+        self.sync_state.mnlistdiff_pipeline.requeue_in_flight();
+    }
+
     async fn handle_message(
         &mut self,
         msg: Message,