feat: implement request timeout handling and tracking for network messages

Author: PastaPastaPasta

dash-spv/src/client/mod.rs

@@ -241,6 +241,10 @@ impl DashSpvClient {
         let mut last_status_update = Instant::now();
         let status_update_interval = std::time::Duration::from_secs(5);
 
+        // Timer for request timeout checking
+        let mut last_timeout_check = Instant::now();
+        let timeout_check_interval = std::time::Duration::from_secs(1);
+        
         loop {
             // Check if we should stop
             let running = self.running.read().await;
@@ -303,6 +307,14 @@ impl DashSpvClient {
             // Check for sync timeouts and handle recovery
             let _ = self.sync_manager.check_sync_timeouts(&mut *self.storage, &mut *self.network).await;
 
+            // Check for request timeouts and handle retries
+            if last_timeout_check.elapsed() >= timeout_check_interval {
+                if let Err(e) = self.handle_request_timeouts().await {
+                    tracing::error!("Error handling request timeouts: {}", e);
+                }
+                last_timeout_check = Instant::now();
+            }
+            
             // Listen for network messages from the per-peer message channel
             match self.network.receive_message().await {
                 Ok(Some(message)) => {
@@ -452,6 +464,11 @@ impl DashSpvClient {
                 let block_hash = block.header.block_hash();
                 tracing::info!("Received new block: {}", block_hash);
                 tracing::debug!("📋 Block {} contains {} transactions", block_hash, block.txdata.len());
+                
+                // Store this as the last successfully received block
+                // This helps identify what block comes next when decoding fails
+                tracing::info!("LAST SUCCESSFUL BLOCK BEFORE POTENTIAL FAILURE: {}", block_hash);
+                
                 // Process new block (update state, check watched items)
                 self.process_new_block(block).await?;
             }
@@ -1541,4 +1558,21 @@ impl DashSpvClient {
         tracing::info!("✅ Completed post-sync filter header requests for {} new blocks", headers.len());
         Ok(())
     }
+    
+    /// Handle request timeouts by calling the network manager's timeout handler.
+    /// This checks for requests that have timed out and initiates retries with exponential backoff.
+    async fn handle_request_timeouts(&mut self) -> Result<()> {
+        // Downcast the network manager to MultiPeerNetworkManager to access request tracker
+        let network = self.network.as_any()
+            .downcast_ref::<crate::network::multi_peer::MultiPeerNetworkManager>()
+            .ok_or_else(|| SpvError::Config("Network manager does not support request tracking".to_string()))?;
+        
+        // Handle timeouts and retries
+        network.handle_request_timeouts().await
+            .map_err(|e| SpvError::Network(crate::error::NetworkError::ConnectionFailed(
+                format!("Failed to handle request timeouts: {}", e)
+            )))?;
+        
+        Ok(())
+    }
 }

dash-spv/src/network/connection.rs

@@ -194,6 +194,13 @@ impl TcpConnection {
 
                 // Message received successfully
                 tracing::trace!("Successfully decoded message from {}: {:?}", self.address, raw_message.payload.cmd());
+                
+                // Log block messages specifically for debugging
+                if let NetworkMessage::Block(ref block) = raw_message.payload {
+                    let block_hash = block.block_hash();
+                    tracing::info!("Successfully decoded block {} from {}", block_hash, self.address);
+                }
+                
                 Ok(Some(raw_message.payload))
             }
             Err(encode::Error::Io(ref e)) if e.kind() == std::io::ErrorKind::WouldBlock => {
@@ -226,8 +233,23 @@ impl TcpConnection {
                 tracing::error!("Failed to decode message from {}: {}", self.address, e);
 
                 // Check if this is the specific "unknown special transaction type" error
-                if e.to_string().contains("unknown special transaction type") {
+                let error_msg = e.to_string();
+                if error_msg.contains("unknown special transaction type") {
                     tracing::warn!("Peer {} sent block with unsupported transaction type: {}", self.address, e);
+                    tracing::error!("BLOCK DECODE FAILURE - Error details: {}", error_msg);
+                } else if error_msg.contains("Failed to decode transactions for block") {
+                    // Extract block hash from the enhanced error message
+                    tracing::error!("Peer {} sent block that failed transaction decoding: {}", self.address, e);
+                    if let Some(hash_start) = error_msg.find("block ") {
+                        if let Some(hash_end) = error_msg[hash_start + 6..].find(':') {
+                            let block_hash = &error_msg[hash_start + 6..hash_start + 6 + hash_end];
+                            tracing::error!("FAILING BLOCK HASH: {}", block_hash);
+                        }
+                    }
+                } else if error_msg.contains("IO error") {
+                    // This might be our wrapped error - log it prominently
+                    tracing::error!("BLOCK DECODE FAILURE - IO error (possibly unknown transaction type) from peer {}", self.address);
+                    tracing::error!("Raw error details: {:?}", e);
                 }
 
                 Err(NetworkError::Serialization(e))

dash-spv/src/network/mod.rs

@@ -10,6 +10,7 @@ pub mod multi_peer;
 pub mod peer;
 pub mod persist;
 pub mod pool;
+pub mod request_tracker;
 
 #[cfg(test)]
 mod tests;