fix(dash-spv): reissue requeued masternode requests from tick

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

@@ -625,6 +625,7 @@ mod tests {
     use dashcore::block::Header;
     use dashcore::hashes::Hash;
     use dashcore::network::message::NetworkMessage;
+    use dashcore::network::message_sml::GetMnListDiff;
     use dashcore::sml::masternode_list::MasternodeList;
     use tokio::sync::mpsc;
 
@@ -1014,4 +1015,52 @@ mod tests {
             "no QRInfo must be requested when the gate picks Incremental"
         );
     }
+
+    /// A single-peer disconnect requeues every in-flight `getmnlistd` request,
+    /// leaving the pipeline with pending work and nothing in flight. The next
+    /// tick must reissue that pending work: a tick that only acts while
+    /// requests are in flight strands the pipeline forever, since every
+    /// response arriving after the requeue is dropped as untracked and no
+    /// other path calls `send_pending`.
+    #[tokio::test]
+    async fn test_tick_reissues_requeued_requests_after_peer_disconnect() {
+        let mut manager = create_test_manager().await;
+        manager.set_state(SyncState::Syncing);
+
+        let base1 = anchor_hash(0x01);
+        let target1 = anchor_hash(0x02);
+        let base2 = anchor_hash(0x03);
+        let target2 = anchor_hash(0x04);
+        let (tx, mut rx) = mpsc::unbounded_channel();
+        let requests = RequestSender::new(tx);
+
+        manager
+            .sync_state
+            .mnlistdiff_pipeline
+            .queue_requests(vec![(base1, target1), (base2, target2)]);
+        manager.sync_state.mnlistdiff_pipeline.send_pending(&requests).unwrap();
+        assert_eq!(manager.sync_state.mnlistdiff_pipeline.active_count(), 2);
+        while rx.try_recv().is_ok() {}
+
+        manager.on_peer_disconnect();
+        assert_eq!(manager.sync_state.mnlistdiff_pipeline.active_count(), 0);
+
+        manager.tick(&requests).await.expect("tick succeeds");
+
+        let mut reissued = Vec::new();
+        while let Ok(request) = rx.try_recv() {
+            if let NetworkRequest::SendMessage(NetworkMessage::GetMnListD(GetMnListDiff {
+                base_block_hash,
+                block_hash,
+            })) = request
+            {
+                reissued.push((base_block_hash, block_hash));
+            }
+        }
+        reissued.sort();
+        let mut expected = vec![(base1, target1), (base2, target2)];
+        expected.sort();
+        assert_eq!(reissued, expected, "tick must reissue the requeued requests");
+        assert_eq!(manager.sync_state.mnlistdiff_pipeline.active_count(), 2);
+    }
 }

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

@@ -164,6 +164,7 @@ impl MnListDiffPipeline {
     }
 
     /// Get the number of in-flight requests.
+    #[cfg(test)]
     pub(super) fn active_count(&self) -> usize {
         self.coordinator.active_count()
     }

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

@@ -627,8 +627,10 @@ impl<H: BlockHeaderStorage> SyncManager for MasternodesManager<H> {
             return Ok(vec![]);
         }
 
-        // Check for MnListDiff timeouts via pipeline
-        if self.sync_state.mnlistdiff_pipeline.active_count() > 0 {
+        // Check for MnListDiff timeouts via pipeline. Gate on outstanding work
+        // rather than in-flight requests: a peer disconnect requeues in-flight
+        // items as pending, and this send is the only path that reissues them.
+        if !self.sync_state.mnlistdiff_pipeline.is_complete() {
             self.sync_state.mnlistdiff_pipeline.handle_timeouts();
 
             // Send any re-queued requests