refactor: Less cloning in SPV message handling (#268)

We currently clone all network messages in `DashSpvClient::handle_network_message` when passed to `MessageHandler::handle_network_message`. Also some get cloned further down the track again. So this PR changes it to pass references.

Author: xdustinface

dash-spv/src/client/message_handler.rs

@@ -60,11 +60,11 @@ impl<
     /// Handle incoming network messages during monitoring.
     pub async fn handle_network_message(
         &mut self,
-        message: dashcore::network::message::NetworkMessage,
+        message: &dashcore::network::message::NetworkMessage,
     ) -> Result<()> {
         use dashcore::network::message::NetworkMessage;
 
-        tracing::debug!("Client handling network message: {:?}", std::mem::discriminant(&message));
+        tracing::debug!("Client handling network message: {:?}", std::mem::discriminant(message));
 
         // First check if this is a message that ONLY the sync manager handles
         // These messages can be moved to the sync manager without cloning
@@ -147,17 +147,11 @@ impl<
                         SpvError::Sync(e)
                     });
             }
-            _ => {}
-        }
-
-        // Handle messages that may need sync manager processing
-        // We optimize to avoid cloning expensive messages like blocks
-        match &message {
             NetworkMessage::Headers(_) | NetworkMessage::CFilter(_) => {
                 // Headers and CFilters are relatively small, cloning is acceptable
                 if let Err(e) = self
                     .sync_manager
-                    .handle_message(message.clone(), &mut *self.network, &mut *self.storage)
+                    .handle_message(message, &mut *self.network, &mut *self.storage)
                     .await
                 {
                     tracing::error!("Sequential sync manager error handling message: {}", e);
@@ -183,13 +177,10 @@ impl<
                 }
             }
             NetworkMessage::Block(_) => {
-                // Blocks can be large - avoid cloning unless necessary
-                // Check if sync manager actually needs to process this block
                 if self.sync_manager.is_in_downloading_blocks_phase() {
-                    // Only clone if we're in the downloading blocks phase
                     if let Err(e) = self
                         .sync_manager
-                        .handle_message(message.clone(), &mut *self.network, &mut *self.storage)
+                        .handle_message(message, &mut *self.network, &mut *self.storage)
                         .await
                     {
                         tracing::error!(
@@ -241,7 +232,7 @@ impl<
                 let headers_msg = NetworkMessage::Headers(vec![block.header]);
                 if let Err(e) = self
                     .sync_manager
-                    .handle_message(headers_msg, &mut *self.network, &mut *self.storage)
+                    .handle_message(&headers_msg, &mut *self.network, &mut *self.storage)
                     .await
                 {
                     tracing::error!(
@@ -261,7 +252,7 @@ impl<
             NetworkMessage::Inv(inv) => {
                 tracing::debug!("Received inventory message with {} items", inv.len());
                 // Handle inventory messages (new blocks, transactions, etc.)
-                self.handle_inventory(inv).await?;
+                self.handle_inventory(inv.clone()).await?;
             }
             NetworkMessage::Tx(tx) => {
                 tracing::info!("📨 Received transaction: {}", tx.txid());
@@ -293,7 +284,7 @@ impl<
                         // Emit event
                         let event = SpvEvent::MempoolTransactionAdded {
                             txid,
-                            transaction: Box::new(tx),
+                            transaction: Box::new(tx.clone()),
                             amount,
                             addresses,
                             is_instant_send,
@@ -339,7 +330,7 @@ impl<
             NetworkMessage::SendDsq(wants_dsq) => {
                 tracing::info!("Received SendDsq message - peer wants DSQ messages: {}", wants_dsq);
                 // Store peer's DSQ preference
-                if let Err(e) = self.network.update_peer_dsq_preference(wants_dsq).await {
+                if let Err(e) = self.network.update_peer_dsq_preference(*wants_dsq).await {
                     tracing::error!("Failed to update peer DSQ preference: {}", e);
                 }
 
@@ -351,7 +342,7 @@ impl<
             }
             _ => {
                 // Ignore other message types for now
-                tracing::debug!("Received network message: {:?}", std::mem::discriminant(&message));
+                tracing::debug!("Received network message: {:?}", std::mem::discriminant(message));
             }
         }
 
@@ -468,7 +459,7 @@ impl<
         // We just need to send them through the unified message interface
         let headers_msg = dashcore::network::message::NetworkMessage::Headers(headers);
         self.sync_manager
-            .handle_message(headers_msg, &mut *self.network, &mut *self.storage)
+            .handle_message(&headers_msg, &mut *self.network, &mut *self.storage)
             .await
             .map_err(SpvError::Sync)?;
 
@@ -506,7 +497,7 @@ impl<
         // For sequential sync, route through the message handler
         let cfheaders_msg = dashcore::network::message::NetworkMessage::CFHeaders(cfheaders);
         self.sync_manager
-            .handle_message(cfheaders_msg, &mut *self.network, &mut *self.storage)
+            .handle_message(&cfheaders_msg, &mut *self.network, &mut *self.storage)
             .await
             .map_err(SpvError::Sync)?;
 
@@ -520,15 +511,15 @@ impl<
     }
 
     /// Process a new block.
-    pub async fn process_new_block(&mut self, block: dashcore::Block) -> Result<()> {
+    pub async fn process_new_block(&mut self, block: &dashcore::Block) -> Result<()> {
         let block_hash = block.block_hash();
 
         tracing::info!("📦 Routing block {} to async block processor", block_hash);
 
         // Send block to the background processor without waiting for completion
         let (response_tx, _response_rx) = tokio::sync::oneshot::channel();
         let task = crate::client::BlockProcessingTask::ProcessBlock {
-            block: Box::new(block),
+            block: Box::new(block.clone()),
             response_tx,
         };
 
@@ -564,7 +555,7 @@ impl<
         // The sequential sync manager's handle_new_headers method will automatically
         // request filter headers and filters as needed
         self.sync_manager
-            .handle_new_headers(headers.to_vec(), &mut *self.network, &mut *self.storage)
+            .handle_new_headers(headers, &mut *self.network, &mut *self.storage)
             .await
             .map_err(SpvError::Sync)?;
 

dash-spv/src/client/sync_coordinator.rs

@@ -633,7 +633,7 @@ impl<
             );
 
             // Delegate message handling to the MessageHandler
-            handler.handle_network_message(message.clone()).await
+            handler.handle_network_message(&message).await
         };
 
         // Handle result and process special messages after releasing storage lock

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

@@ -246,7 +246,7 @@ impl<S: StorageManager + Send + Sync + 'static, N: NetworkManager + Send + Sync
 
     pub async fn store_filter_headers(
         &mut self,
-        cfheaders: dashcore::network::message_filter::CFHeaders,
+        cfheaders: &dashcore::network::message_filter::CFHeaders,
         storage: &mut S,
     ) -> SyncResult<()> {
         if cfheaders.filter_hashes.is_empty() {
@@ -256,7 +256,7 @@ impl<S: StorageManager + Send + Sync + 'static, N: NetworkManager + Send + Sync
 
         // Get the height range for this batch
         let (start_height, stop_height, _header_tip_height) =
-            self.get_batch_height_range(&cfheaders, storage).await?;
+            self.get_batch_height_range(cfheaders, storage).await?;
 
         tracing::info!(
             "Received {} filter headers from height {} to {}",
@@ -308,7 +308,7 @@ impl<S: StorageManager + Send + Sync + 'static, N: NetworkManager + Send + Sync
             // Use the handle_overlapping_headers method which properly handles the chain continuity
             let expected_start = current_filter_tip + 1;
 
-            match self.handle_overlapping_headers(&cfheaders, expected_start, storage).await {
+            match self.handle_overlapping_headers(cfheaders, expected_start, storage).await {
                 Ok((stored_count, _)) => {
                     if stored_count > 0 {
                         tracing::info!("✅ Successfully handled overlapping filter headers");
@@ -328,7 +328,7 @@ impl<S: StorageManager + Send + Sync + 'static, N: NetworkManager + Send + Sync
             }
         } else {
             // Process the filter headers to convert them to the proper format
-            match self.process_filter_headers(&cfheaders, start_height, storage).await {
+            match self.process_filter_headers(cfheaders, start_height, storage).await {
                 Ok(new_filter_headers) => {
                     if !new_filter_headers.is_empty() {
                         // If this is the first batch (starting at height 1), store the genesis filter header first

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

@@ -671,7 +671,7 @@ impl<S: StorageManager + Send + Sync + 'static, N: NetworkManager + Send + Sync
                 );
 
                 // Store the verified filter headers
-                self.store_filter_headers(cf_headers.clone(), storage).await?;
+                self.store_filter_headers(&cf_headers, storage).await?;
 
                 // Update next expected height
                 self.next_cfheader_height_to_process = stop_height + 1;

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

@@ -139,7 +139,7 @@ impl<S: StorageManager + Send + Sync + 'static, N: NetworkManager + Send + Sync
     /// Handle a Headers message
     pub async fn handle_headers_message(
         &mut self,
-        headers: Vec<BlockHeader>,
+        headers: &[BlockHeader],
         storage: &mut S,
         network: &mut N,
     ) -> SyncResult<bool> {
@@ -251,7 +251,7 @@ impl<S: StorageManager + Send + Sync + 'static, N: NetworkManager + Send + Sync
         // Update Chain State: Add all headers to in-memory chain_state
         {
             let mut cs = self.chain_state.write().await;
-            for header in &headers {
+            for header in headers {
                 cs.add_header(*header);
             }
         }
@@ -261,13 +261,13 @@ impl<S: StorageManager + Send + Sync + 'static, N: NetworkManager + Send + Sync
             storage.as_any_mut().downcast_mut::<crate::storage::disk::DiskStorageManager>()
         {
             disk_storage
-                .store_headers(&headers)
+                .store_headers(headers)
                 .await
                 .map_err(|e| SyncError::Storage(format!("Failed to store headers batch: {}", e)))?;
         } else {
             // Fallback to standard store_headers for other storage backends
             storage
-                .store_headers(&headers)
+                .store_headers(headers)
                 .await
                 .map_err(|e| SyncError::Storage(format!("Failed to store headers batch: {}", e)))?;
         }
@@ -452,7 +452,7 @@ impl<S: StorageManager + Send + Sync + 'static, N: NetworkManager + Send + Sync
     /// Returns true if the message was processed and sync should continue, false if sync is complete.
     pub async fn handle_headers2_message(
         &mut self,
-        headers2: dashcore::network::message_headers2::Headers2Message,
+        headers2: &dashcore::network::message_headers2::Headers2Message,
         peer_id: crate::types::PeerId,
         _storage: &mut S,
         _network: &mut N,
@@ -553,7 +553,7 @@ impl<S: StorageManager + Send + Sync + 'static, N: NetworkManager + Send + Sync
             );
 
             // Process decompressed headers through the normal flow
-            self.handle_headers_message(headers, _storage, _network).await
+            self.handle_headers_message(&headers, _storage, _network).await
         }
     }
 

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

@@ -443,11 +443,11 @@ impl<S: StorageManager + Send + Sync + 'static, N: NetworkManager + Send + Sync
     /// Handle incoming MnListDiff message
     pub async fn handle_mnlistdiff_message(
         &mut self,
-        diff: MnListDiff,
+        diff: &MnListDiff,
         storage: &mut S,
         _network: &mut dyn NetworkManager,
     ) -> SyncResult<bool> {
-        self.insert_mn_list_diff(&diff, storage).await;
+        self.insert_mn_list_diff(diff, storage).await;
 
         // Decrement pending request counter if we were expecting this response
         if self.pending_mnlistdiff_requests > 0 {

dash-spv/src/sync/message_handlers.rs

@@ -25,7 +25,7 @@ impl<
     /// Handle incoming network messages with phase filtering
     pub async fn handle_message(
         &mut self,
-        message: NetworkMessage,
+        message: &NetworkMessage,
         network: &mut N,
         storage: &mut S,
     ) -> SyncResult<()> {
@@ -55,10 +55,10 @@ impl<
         }
 
         // Check if this message is expected in the current phase
-        if !self.is_message_expected_in_phase(&message) {
+        if !self.is_message_expected_in_phase(message) {
             tracing::debug!(
                 "Ignoring unexpected {:?} message in phase {}",
-                std::mem::discriminant(&message),
+                std::mem::discriminant(message),
                 self.current_phase.name()
             );
             return Ok(());
@@ -286,7 +286,7 @@ impl<
 
     pub(super) async fn handle_headers2_message(
         &mut self,
-        headers2: dashcore::network::message_headers2::Headers2Message,
+        headers2: &dashcore::network::message_headers2::Headers2Message,
         peer_id: PeerId,
         network: &mut N,
         storage: &mut S,
@@ -345,12 +345,12 @@ impl<
 
     pub(super) async fn handle_headers_message(
         &mut self,
-        headers: Vec<dashcore::block::Header>,
+        headers: &[dashcore::block::Header],
         network: &mut N,
         storage: &mut S,
     ) -> SyncResult<()> {
         let continue_sync =
-            self.header_sync.handle_headers_message(headers.clone(), storage, network).await?;
+            self.header_sync.handle_headers_message(headers, storage, network).await?;
 
         // Calculate blockchain height before borrowing self.current_phase
         let blockchain_height = self.get_blockchain_height_from_storage(storage).await.unwrap_or(0);
@@ -400,7 +400,7 @@ impl<
 
     pub(super) async fn handle_mnlistdiff_message(
         &mut self,
-        diff: dashcore::network::message_sml::MnListDiff,
+        diff: &dashcore::network::message_sml::MnListDiff,
         network: &mut N,
         storage: &mut S,
     ) -> SyncResult<()> {
@@ -451,7 +451,7 @@ impl<
 
     pub(super) async fn handle_qrinfo_message(
         &mut self,
-        qr_info: dashcore::network::message_qrinfo::QRInfo,
+        qr_info: &dashcore::network::message_qrinfo::QRInfo,
         network: &mut N,
         storage: &mut S,
     ) -> SyncResult<()> {
@@ -509,7 +509,7 @@ impl<
 
     pub(super) async fn handle_cfheaders_message(
         &mut self,
-        cfheaders: dashcore::network::message_filter::CFHeaders,
+        cfheaders: &dashcore::network::message_filter::CFHeaders,
         network: &mut N,
         storage: &mut S,
     ) -> SyncResult<()> {
@@ -563,7 +563,7 @@ impl<
 
     pub(super) async fn handle_cfilter_message(
         &mut self,
-        cfilter: dashcore::network::message_filter::CFilter,
+        cfilter: &dashcore::network::message_filter::CFilter,
         network: &mut N,
         storage: &mut S,
     ) -> SyncResult<()> {
@@ -752,7 +752,7 @@ impl<
 
     pub(super) async fn handle_block_message(
         &mut self,
-        block: Block,
+        block: &Block,
         network: &mut N,
         storage: &mut S,
     ) -> SyncResult<()> {
@@ -768,7 +768,7 @@ impl<
             .map_err(|e| SyncError::Storage(format!("Failed to get block height: {}", e)))?
             .unwrap_or(0);
 
-        let relevant_txids = wallet.process_block(&block, block_height, self.config.network).await;
+        let relevant_txids = wallet.process_block(block, block_height, self.config.network).await;
 
         drop(wallet);