feat: various notification / statistics and performance improvements

dash-spv/src/client/block_processor.rs

@@ -234,6 +234,39 @@ impl<W: WalletInterface + Send + Sync + 'static, S: StorageManager + Send + Sync
                 block_hash,
                 height
             );
+
+            // Update statistics for blocks with relevant transactions
+            {
+                let mut stats = self.stats.write().await;
+                stats.blocks_with_relevant_transactions += 1;
+            }
+
+            // Emit TransactionDetected events for each relevant transaction
+            for txid in &txids {
+                if let Some(tx) = block.txdata.iter().find(|t| &t.txid() == txid) {
+                    // Ask the wallet for the precise effect of this transaction
+                    let effect = wallet.transaction_effect(tx, self.network).await;
+                    if let Some((net_amount, affected_addresses)) = effect {
+                        tracing::info!("📤 Emitting TransactionDetected event for {}", txid);
+                        let _ = self.event_tx.send(SpvEvent::TransactionDetected {
+                            txid: txid.to_string(),
+                            confirmed: true,
+                            block_height: Some(height),
+                            amount: net_amount,
+                            addresses: affected_addresses,
+                        });
+                    } else {
+                        // Fallback: emit event with zero and no addresses if wallet could not compute
+                        let _ = self.event_tx.send(SpvEvent::TransactionDetected {
+                            txid: txid.to_string(),
+                            confirmed: true,
+                            block_height: Some(height),
+                            amount: 0,
+                            addresses: Vec::new(),
+                        });
+                    }
+                }
+            }
         }
         drop(wallet); // Release lock
 

dash-spv/src/client/block_processor_test.rs

@@ -12,19 +12,31 @@ mod tests {
     use std::sync::Arc;
     use tokio::sync::{mpsc, oneshot, Mutex, RwLock};
 
+    // Type alias for transaction effects map
+    type TransactionEffectsMap =
+        Arc<Mutex<std::collections::BTreeMap<dashcore::Txid, (i64, Vec<String>)>>>;
+
     // Mock WalletInterface implementation for testing
     struct MockWallet {
         processed_blocks: Arc<Mutex<Vec<(dashcore::BlockHash, u32)>>>,
         processed_transactions: Arc<Mutex<Vec<dashcore::Txid>>>,
+        // Map txid -> (net_amount, addresses)
+        effects: TransactionEffectsMap,
     }
 
     impl MockWallet {
         fn new() -> Self {
             Self {
                 processed_blocks: Arc::new(Mutex::new(Vec::new())),
                 processed_transactions: Arc::new(Mutex::new(Vec::new())),
+                effects: Arc::new(Mutex::new(std::collections::BTreeMap::new())),
             }
         }
+
+        async fn set_effect(&self, txid: dashcore::Txid, net: i64, addresses: Vec<String>) {
+            let mut map = self.effects.lock().await;
+            map.insert(txid, (net, addresses));
+        }
     }
 
     #[async_trait::async_trait]
@@ -64,6 +76,15 @@ mod tests {
         async fn describe(&self, _network: Network) -> String {
             "MockWallet (test implementation)".to_string()
         }
+
+        async fn transaction_effect(
+            &self,
+            tx: &Transaction,
+            _network: Network,
+        ) -> Option<(i64, Vec<String>)> {
+            let map = self.effects.lock().await;
+            map.get(&tx.txid()).cloned()
+        }
     }
 
     fn create_test_block(network: Network) -> Block {
@@ -110,6 +131,15 @@ mod tests {
 
         // Send block processing task
         let (response_tx, _response_rx) = oneshot::channel();
+
+        // Prime wallet with an effect for the coinbase tx in the genesis block
+        let txid = block.txdata[0].txid();
+        {
+            let wallet_guard = wallet.read().await;
+            wallet_guard
+                .set_effect(txid, 1234, vec!["XyTestAddr1".to_string(), "XyTestAddr2".to_string()])
+                .await;
+        }
         task_tx
             .send(BlockProcessingTask::ProcessBlock {
                 block: Box::new(block.clone()),
@@ -120,22 +150,47 @@ mod tests {
         // Process the block in a separate task
         let processor_handle = tokio::spawn(async move { processor.run().await });
 
-        // Wait for event
+        // Wait for events; capture the TransactionDetected for our tx
+        let mut saw_tx_event = false;
         tokio::time::timeout(std::time::Duration::from_millis(100), async {
             while let Some(event) = event_rx.recv().await {
-                if let SpvEvent::BlockProcessed {
-                    hash,
-                    ..
-                } = event
-                {
-                    assert_eq!(hash.to_string(), block_hash.to_string());
-                    break;
+                match event {
+                    SpvEvent::TransactionDetected {
+                        txid: tid,
+                        amount,
+                        addresses,
+                        confirmed,
+                        block_height,
+                    } => {
+                        // Should use wallet-provided values
+                        assert_eq!(tid, txid.to_string());
+                        assert_eq!(amount, 1234);
+                        assert_eq!(
+                            addresses,
+                            vec!["XyTestAddr1".to_string(), "XyTestAddr2".to_string()]
+                        );
+                        assert!(confirmed);
+                        assert_eq!(block_height, Some(0));
+                        saw_tx_event = true;
+                    }
+                    SpvEvent::BlockProcessed {
+                        hash,
+                        ..
+                    } => {
+                        assert_eq!(hash.to_string(), block_hash.to_string());
+                        if saw_tx_event {
+                            break;
+                        }
+                    }
+                    _ => {}
                 }
             }
         })
         .await
         .expect("Should receive block processed event");
 
+        assert!(saw_tx_event, "Should emit TransactionDetected with wallet-provided effect");
+
         // Verify wallet was called
         {
             let wallet = wallet.read().await;
@@ -292,6 +347,158 @@ mod tests {
         let _ = processor_handle.await;
     }
 
+    #[tokio::test]
+    async fn test_transaction_detected_fallback_when_no_wallet_effect() {
+        let (processor, task_tx, mut event_rx, _wallet, storage) = setup_processor().await;
+
+        // Create a test block
+        let block = create_test_block(Network::Dash);
+        let block_hash = block.block_hash();
+        let txid = block.txdata[0].txid();
+
+        // Store header so height lookup succeeds
+        {
+            let mut storage = storage.lock().await;
+            storage.store_headers(&[block.header]).await.unwrap();
+        }
+
+        // Send block processing task without priming any effect (transaction_effect will return None)
+        let (response_tx, _response_rx) = oneshot::channel();
+        task_tx
+            .send(BlockProcessingTask::ProcessBlock {
+                block: Box::new(block.clone()),
+                response_tx,
+            })
+            .unwrap();
+
+        // Process
+        let processor_handle = tokio::spawn(async move { processor.run().await });
+
+        let mut saw_tx_event = false;
+        tokio::time::timeout(std::time::Duration::from_millis(100), async {
+            while let Some(event) = event_rx.recv().await {
+                match event {
+                    SpvEvent::TransactionDetected {
+                        txid: tid,
+                        amount,
+                        addresses,
+                        confirmed,
+                        block_height,
+                    } => {
+                        assert_eq!(tid, txid.to_string());
+                        assert_eq!(
+                            amount, 0,
+                            "fallback amount should be 0 when no effect available"
+                        );
+                        assert!(addresses.is_empty(), "fallback addresses should be empty");
+                        assert!(confirmed);
+                        assert_eq!(block_height, Some(0));
+                        saw_tx_event = true;
+                    }
+                    SpvEvent::BlockProcessed {
+                        hash,
+                        ..
+                    } => {
+                        assert_eq!(hash.to_string(), block_hash.to_string());
+                        if saw_tx_event {
+                            break;
+                        }
+                    }
+                    _ => {}
+                }
+            }
+        })
+        .await
+        .expect("Should receive events");
+
+        assert!(saw_tx_event, "Should emit TransactionDetected with fallback values");
+
+        // Shutdown
+        drop(task_tx);
+        let _ = processor_handle.await;
+    }
+
+    #[tokio::test]
+    async fn test_transaction_detected_negative_amount_and_duplicate_addresses() {
+        let (processor, task_tx, mut event_rx, wallet, storage) = setup_processor().await;
+
+        // Create a test block
+        let block = create_test_block(Network::Dash);
+        let block_hash = block.block_hash();
+        let txid = block.txdata[0].txid();
+
+        // Store header so height lookup succeeds
+        {
+            let mut storage = storage.lock().await;
+            storage.store_headers(&[block.header]).await.unwrap();
+        }
+
+        // Prime wallet with negative amount and duplicate addresses
+        {
+            let wallet_guard = wallet.read().await;
+            wallet_guard
+                .set_effect(
+                    txid,
+                    -500,
+                    vec!["DupAddr".to_string(), "DupAddr".to_string(), "UniqueAddr".to_string()],
+                )
+                .await;
+        }
+
+        // Send block processing task
+        let (response_tx, _response_rx) = oneshot::channel();
+        task_tx
+            .send(BlockProcessingTask::ProcessBlock {
+                block: Box::new(block.clone()),
+                response_tx,
+            })
+            .unwrap();
+
+        // Process
+        let processor_handle = tokio::spawn(async move { processor.run().await });
+
+        let mut saw_tx_event = false;
+        tokio::time::timeout(std::time::Duration::from_millis(100), async {
+            while let Some(event) = event_rx.recv().await {
+                match event {
+                    SpvEvent::TransactionDetected {
+                        txid: tid,
+                        amount,
+                        addresses,
+                        confirmed,
+                        block_height,
+                    } => {
+                        assert_eq!(tid, txid.to_string());
+                        assert_eq!(amount, -500);
+                        // BlockProcessor uses wallet-provided addresses as-is (no dedup here)
+                        assert_eq!(addresses, vec!["DupAddr", "DupAddr", "UniqueAddr"]);
+                        assert!(confirmed);
+                        assert_eq!(block_height, Some(0));
+                        saw_tx_event = true;
+                    }
+                    SpvEvent::BlockProcessed {
+                        hash,
+                        ..
+                    } => {
+                        assert_eq!(hash.to_string(), block_hash.to_string());
+                        if saw_tx_event {
+                            break;
+                        }
+                    }
+                    _ => {}
+                }
+            }
+        })
+        .await
+        .expect("Should receive events");
+
+        assert!(saw_tx_event, "Should emit TransactionDetected with negative net and duplicates");
+
+        // Shutdown
+        drop(task_tx);
+        let _ = processor_handle.await;
+    }
+
     #[tokio::test]
     async fn test_process_mempool_transaction() {
         let (processor, task_tx, _event_rx, wallet, _storage) = setup_processor().await;

dash-spv/src/client/config.rs

@@ -163,6 +163,19 @@ pub struct ClientConfig {
     /// Rate limit for CF header requests per second (default: 10.0).
     pub cfheaders_request_rate_limit: Option<f64>,
 
+    // CFHeaders flow control configuration
+    /// Maximum concurrent CFHeaders requests for parallel sync (default: 50).
+    pub max_concurrent_cfheaders_requests_parallel: usize,
+
+    /// Enable flow control for CFHeaders requests (default: true).
+    pub enable_cfheaders_flow_control: bool,
+
+    /// Timeout for CFHeaders requests in seconds (default: 30).
+    pub cfheaders_request_timeout_secs: u64,
+
+    /// Maximum retry attempts for failed CFHeaders batches (default: 3).
+    pub max_cfheaders_retries: u32,
+
     /// Rate limit for filter requests per second (default: 50.0).
     pub filters_request_rate_limit: Option<f64>,
 
@@ -238,6 +251,11 @@ impl Default for ClientConfig {
             blocks_request_rate_limit: None,
             start_from_height: None,
             wallet_creation_time: None,
+            // CFHeaders flow control defaults
+            max_concurrent_cfheaders_requests_parallel: 50,
+            enable_cfheaders_flow_control: true,
+            cfheaders_request_timeout_secs: 30,
+            max_cfheaders_retries: 3,
             // QRInfo defaults (simplified per plan)
             qr_info_extra_share: false, // Matches DMLviewer.patch default
             qr_info_timeout: Duration::from_secs(30),

dash-spv/src/client/mod.rs

@@ -311,6 +311,7 @@ impl<
             received_filter_heights,
             wallet.clone(),
             state.clone(),
+            stats.clone(),
         )
         .map_err(SpvError::Sync)?;
 

dash-spv/src/network/tests.rs

@@ -146,6 +146,10 @@ mod multi_peer_tests {
             enable_filter_flow_control: true,
             filter_request_delay_ms: 0,
             max_concurrent_filter_requests: 50,
+            max_concurrent_cfheaders_requests_parallel: 50,
+            enable_cfheaders_flow_control: true,
+            cfheaders_request_timeout_secs: 30,
+            max_cfheaders_retries: 3,
             enable_cfheader_gap_restart: true,
             cfheader_gap_check_interval_secs: 15,
             cfheader_gap_restart_cooldown_secs: 30,