diff --git a/dash-spv/src/sync/block_headers/sync_manager.rs b/dash-spv/src/sync/block_headers/sync_manager.rs index d47e2cfe8..600ee58e8 100644 --- a/dash-spv/src/sync/block_headers/sync_manager.rs +++ b/dash-spv/src/sync/block_headers/sync_manager.rs @@ -47,6 +47,17 @@ impl 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> { ensure_not_started(self.state(), self.identifier())?; self.progress.set_state(SyncState::Syncing); @@ -191,6 +202,14 @@ impl 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, diff --git a/dash-spv/src/sync/blocks/manager.rs b/dash-spv/src/sync/blocks/manager.rs index 70df8624b..feb428da6 100644 --- a/dash-spv/src/sync/blocks/manager.rs +++ b/dash-spv/src/sync/blocks/manager.rs @@ -167,16 +167,20 @@ impl 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; @@ -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> = (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, + ) -> BTreeSet { + 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. diff --git a/dash-spv/src/sync/blocks/sync_manager.rs b/dash-spv/src/sync/blocks/sync_manager.rs index e7ecbc68a..9eeeda96d 100644 --- a/dash-spv/src/sync/blocks/sync_manager.rs +++ b/dash-spv/src/sync/blocks/sync_manager.rs @@ -62,6 +62,10 @@ impl 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, diff --git a/dash-spv/src/sync/download_coordinator.rs b/dash-spv/src/sync/download_coordinator.rs index e36753b6d..6acb0f3ca 100644 --- a/dash-spv/src/sync/download_coordinator.rs +++ b/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] diff --git a/dash-spv/src/sync/filter_headers/pipeline.rs b/dash-spv/src/sync/filter_headers/pipeline.rs index 309b28ca0..97911cbad 100644 --- a/dash-spv/src/sync/filter_headers/pipeline.rs +++ b/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; diff --git a/dash-spv/src/sync/filter_headers/sync_manager.rs b/dash-spv/src/sync/filter_headers/sync_manager.rs index eae554d57..d33d70864 100644 --- a/dash-spv/src/sync/filter_headers/sync_manager.rs +++ b/dash-spv/src/sync/filter_headers/sync_manager.rs @@ -37,6 +37,10 @@ impl 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, diff --git a/dash-spv/src/sync/filters/sync_manager.rs b/dash-spv/src/sync/filters/sync_manager.rs index 81bec92d4..6cc9d7a9e 100644 --- a/dash-spv/src/sync/filters/sync_manager.rs +++ b/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> { ensure_not_started(self.state(), self.identifier())?; diff --git a/dash-spv/src/sync/masternodes/pipeline.rs b/dash-spv/src/sync/masternodes/pipeline.rs index 9a67149ec..02d50ca67 100644 --- a/dash-spv/src/sync/masternodes/pipeline.rs +++ b/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(); diff --git a/dash-spv/src/sync/masternodes/sync_manager.rs b/dash-spv/src/sync/masternodes/sync_manager.rs index 69c10209a..1d17cfa6d 100644 --- a/dash-spv/src/sync/masternodes/sync_manager.rs +++ b/dash-spv/src/sync/masternodes/sync_manager.rs @@ -231,6 +231,12 @@ impl SyncManager for MasternodesManager { 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, diff --git a/dash-spv/src/sync/sync_manager.rs b/dash-spv/src/sync/sync_manager.rs index 997e9f505..f27ffeaff 100644 --- a/dash-spv/src/sync/sync_manager.rs +++ b/dash-spv/src/sync/sync_manager.rs @@ -133,6 +133,20 @@ pub trait SyncManager: Send + Sync + std::fmt::Debug { /// immediately to the new peer. fn on_disconnect(&mut self); + /// Requeue in-flight work after a single peer drops while others remain. + /// + /// Distinct from [`SyncManager::on_disconnect`], which runs only once every + /// peer is gone and is therefore free to discard peer-bound state wholesale. + /// Here the surviving peers can still serve the work, so an implementation + /// must requeue and nothing else. + /// + /// In-flight items carry no peer attribution, so an implementation requeues + /// everything outstanding, including requests a healthy peer is still going + /// to answer. Retry counts survive a requeue and every receive path treats a + /// response it no longer tracks as unrequested, so the cost is redundant + /// traffic rather than lost work or a corrupted retry budget. + fn on_peer_disconnect(&mut self) {} + /// Handle an incoming network message. /// /// Returns events to emit to other managers. @@ -171,27 +185,38 @@ pub trait SyncManager: Send + Sync + std::fmt::Debug { event: &NetworkEvent, requests: &RequestSender, ) -> SyncResult> { - // Default: transition from WaitingForConnections to Syncing when peers connect - if let NetworkEvent::PeersUpdated { - connected_count, - best_height, - .. - } = event - { - if let Some(best_height) = best_height { - self.update_target_height(*best_height); - } - if *connected_count == 0 { - tracing::info!("{} - no peers available, stopping sync", self.identifier()); - self.stop_sync(); - } else if *connected_count > 0 && self.state() == SyncState::WaitingForConnections { - tracing::info!( - "{} - peers available ({}), starting sync", - self.identifier(), - connected_count - ); - return self.start_sync(requests).await; + match event { + // `PeersUpdated` carries only the surviving count, so a drop that + // leaves other peers connected is invisible there. `PeerDisconnected` + // precedes it for every removal path and is what makes the drop + // observable at all. + NetworkEvent::PeerDisconnected { + .. + } => self.on_peer_disconnect(), + // Default: transition from WaitingForConnections to Syncing when peers connect + NetworkEvent::PeersUpdated { + connected_count, + best_height, + .. + } => { + if let Some(best_height) = best_height { + self.update_target_height(*best_height); + } + if *connected_count == 0 { + tracing::info!("{} - no peers available, stopping sync", self.identifier()); + self.stop_sync(); + } else if *connected_count > 0 && self.state() == SyncState::WaitingForConnections { + tracing::info!( + "{} - peers available ({}), starting sync", + self.identifier(), + connected_count + ); + return self.start_sync(requests).await; + } } + NetworkEvent::PeerConnected { + .. + } => {} } Ok(vec![]) } @@ -340,6 +365,7 @@ mod tests { use crate::network::NetworkRequest; use crate::sync::BlockHeadersProgress; use crate::sync::SyncState; + use crate::test_utils::test_socket_address; use async_trait::async_trait; use std::sync::atomic::{AtomicU32, Ordering}; use std::sync::Arc; @@ -352,6 +378,24 @@ mod tests { message_count: Arc, event_count: Arc, tick_count: Arc, + /// Stands in for the destructive state a total-loss `on_disconnect` + /// throws away. + progress_kept: bool, + requeue_count: u32, + } + + impl MockManager { + fn new(state: SyncState) -> Self { + Self { + identifier: ManagerIdentifier::BlockHeader, + state, + message_count: Arc::new(AtomicU32::new(0)), + event_count: Arc::new(AtomicU32::new(0)), + tick_count: Arc::new(AtomicU32::new(0)), + progress_kept: true, + requeue_count: 0, + } + } } impl std::fmt::Debug for MockManager { @@ -378,7 +422,13 @@ mod tests { &[] } - fn on_disconnect(&mut self) {} + fn on_disconnect(&mut self) { + self.progress_kept = false; + } + + fn on_peer_disconnect(&mut self) { + self.requeue_count += 1; + } async fn handle_message( &mut self, @@ -412,17 +462,8 @@ mod tests { #[tokio::test] async fn test_manager_task_shutdown() { - let message_count = Arc::new(AtomicU32::new(0)); - let event_count = Arc::new(AtomicU32::new(0)); - let tick_count = Arc::new(AtomicU32::new(0)); - - let manager = MockManager { - identifier: ManagerIdentifier::BlockHeader, - state: SyncState::WaitForEvents, - message_count: message_count.clone(), - event_count: event_count.clone(), - tick_count: tick_count.clone(), - }; + let manager = MockManager::new(SyncState::WaitForEvents); + let tick_count = manager.tick_count.clone(); // Create channels let (_, message_receiver) = mpsc::unbounded_channel(); @@ -461,4 +502,43 @@ mod tests { // Verify tick was called multiple times assert!(tick_count.load(Ordering::Relaxed) > 0); } + + /// Losing one peer of several must reach the requeue hook and leave the + /// destructive total-loss hook alone. Only the final `PeersUpdated` with a + /// zero count is allowed to discard progress. + #[tokio::test] + async fn test_peer_disconnect_requeues_without_dropping_progress() { + let mut manager = MockManager::new(SyncState::Syncing); + let (req_tx, _req_rx) = mpsc::unbounded_channel::(); + let requests = RequestSender::new(req_tx); + + let disconnect = NetworkEvent::PeerDisconnected { + address: test_socket_address(1), + }; + let survivors = NetworkEvent::PeersUpdated { + connected_count: 2, + addresses: vec![test_socket_address(2), test_socket_address(3)], + best_height: Some(1000), + }; + + manager.handle_network_event(&disconnect, &requests).await.unwrap(); + manager.handle_network_event(&survivors, &requests).await.unwrap(); + + assert_eq!(manager.requeue_count, 1); + assert!(manager.progress_kept); + assert_eq!(manager.state(), SyncState::Syncing); + + // The last peer going away still runs the destructive hook. + let no_peers = NetworkEvent::PeersUpdated { + connected_count: 0, + addresses: vec![], + best_height: Some(1000), + }; + manager.handle_network_event(&disconnect, &requests).await.unwrap(); + manager.handle_network_event(&no_peers, &requests).await.unwrap(); + + assert_eq!(manager.requeue_count, 2); + assert!(!manager.progress_kept); + assert_eq!(manager.state(), SyncState::WaitingForConnections); + } }