refactor: migrate Mutex to tokio::sync::Mutex for async compatibility

- Updated `received_filter_heights` and `processing_thread_requests` to use `tokio::sync::Mutex` instead of `std::sync::Mutex` for better async support.
- Adjusted related locking mechanisms in `FilterSyncManager`, `SpvStats`, and various test files to accommodate the new async locking.
- Enhanced test files to reflect the changes in synchronization primitives, ensuring compatibility with async operations.

Author: PastaPastaPasta

dash-spv-ffi/tests/unit/test_type_conversions.rs

@@ -209,7 +209,7 @@ mod tests {
             filters_received: u64::MAX,
             filter_sync_start_time: None,
             last_filter_received_time: None,
-            received_filter_heights: std::sync::Arc::new(std::sync::Mutex::new(
+            received_filter_heights: std::sync::Arc::new(tokio::sync::Mutex::new(
                 std::collections::HashSet::new(),
             )),
             active_filter_requests: 0,

dash-spv/src/client/status_display.rs

@@ -76,10 +76,9 @@ impl<'a, S: StorageManager + Send + Sync + 'static> StatusDisplay<'a, S> {
         let stats = self.stats.read().await;
 
         // Calculate last synced filter height from received filter heights
-        let last_synced_filter_height = if let Ok(heights) = stats.received_filter_heights.lock() {
+        let last_synced_filter_height = {
+            let heights = stats.received_filter_heights.lock().await;
             heights.iter().max().copied()
-        } else {
-            None
         };
 
         // Calculate the actual header height considering checkpoint sync

dash-spv/src/sync/filters.rs

@@ -74,11 +74,11 @@ pub struct FilterSyncManager<S: StorageManager, N: NetworkManager> {
     /// Blocks currently being downloaded (map for quick lookup)
     downloading_blocks: HashMap<BlockHash, u32>,
     /// Blocks requested by the filter processing thread
-    pub processing_thread_requests: std::sync::Arc<std::sync::Mutex<HashSet<BlockHash>>>,
+    pub processing_thread_requests: std::sync::Arc<tokio::sync::Mutex<HashSet<BlockHash>>>,
     /// Track requested filter ranges: (start_height, end_height) -> request_time
     requested_filter_ranges: HashMap<(u32, u32), std::time::Instant>,
     /// Track individual filter heights that have been received (shared with stats)
-    received_filter_heights: std::sync::Arc<std::sync::Mutex<HashSet<u32>>>,
+    received_filter_heights: std::sync::Arc<tokio::sync::Mutex<HashSet<u32>>>,
     /// Maximum retries for a filter range
     max_filter_retries: u32,
     /// Retry attempts per range
@@ -236,7 +236,7 @@ impl<S: StorageManager + Send + Sync + 'static, N: NetworkManager + Send + Sync
     /// Create a new filter sync manager.
     pub fn new(
         config: &ClientConfig,
-        received_filter_heights: std::sync::Arc<std::sync::Mutex<std::collections::HashSet<u32>>>,
+        received_filter_heights: std::sync::Arc<tokio::sync::Mutex<std::collections::HashSet<u32>>>,
     ) -> Self {
         Self {
             _config: config.clone(),
@@ -249,7 +249,7 @@ impl<S: StorageManager + Send + Sync + 'static, N: NetworkManager + Send + Sync
             syncing_filters: false,
             pending_block_downloads: VecDeque::new(),
             downloading_blocks: HashMap::new(),
-            processing_thread_requests: std::sync::Arc::new(std::sync::Mutex::new(
+            processing_thread_requests: std::sync::Arc::new(tokio::sync::Mutex::new(
                 std::collections::HashSet::new(),
             )),
             requested_filter_ranges: HashMap::new(),
@@ -1698,7 +1698,8 @@ impl<S: StorageManager + Send + Sync + 'static, N: NetworkManager + Send + Sync
         }
 
         // Log current state periodically
-        if let Ok(guard) = self.received_filter_heights.lock() {
+        {
+            let guard = self.received_filter_heights.lock().await;
             if guard.len() % 1000 == 0 {
                 tracing::info!(
                     "Filter sync state: {} filters received, {} active requests, {} pending requests",
@@ -1725,16 +1726,13 @@ impl<S: StorageManager + Send + Sync + 'static, N: NetworkManager + Send + Sync
 
     /// Check if a filter request range is complete (all filters received).
     async fn is_request_complete(&self, start: u32, end: u32) -> SyncResult<bool> {
-        if let Ok(received_heights) = self.received_filter_heights.lock() {
-            for height in start..=end {
-                if !received_heights.contains(&height) {
-                    return Ok(false);
-                }
+        let received_heights = self.received_filter_heights.lock().await;
+        for height in start..=end {
+            if !received_heights.contains(&height) {
+                return Ok(false);
             }
-            Ok(true)
-        } else {
-            Err(SyncError::Storage("Failed to lock received filter heights".to_string()))
         }
+        Ok(true)
     }
 
     /// Record that a filter was received at a specific height.
@@ -1748,14 +1746,13 @@ impl<S: StorageManager + Send + Sync + 'static, N: NetworkManager + Send + Sync
             SyncError::Storage(format!("Failed to get header height by hash: {}", e))
         })? {
             // Record in received filter heights
-            if let Ok(mut heights) = self.received_filter_heights.lock() {
-                heights.insert(height);
-                tracing::trace!(
-                    "📊 Recorded filter received at height {} for block {}",
-                    height,
-                    block_hash
-                );
-            }
+            let mut heights = self.received_filter_heights.lock().await;
+            heights.insert(height);
+            tracing::trace!(
+                "📊 Recorded filter received at height {} for block {}",
+                height,
+                block_hash
+            );
         } else {
             tracing::warn!("Could not find height for filter block hash {}", block_hash);
         }
@@ -2429,9 +2426,7 @@ impl<S: StorageManager + Send + Sync + 'static, N: NetworkManager + Send + Sync
 
         // Check if this block was requested by the filter processing thread
         {
-            let mut processing_requests = self.processing_thread_requests.lock().map_err(|e| {
-                SyncError::InvalidState(format!("processing thread requests lock poisoned: {}", e))
-            })?;
+            let mut processing_requests = self.processing_thread_requests.lock().await;
             if processing_requests.remove(&block_hash) {
                 tracing::info!(
                     "📦 Received block {} requested by filter processing thread",
@@ -2607,10 +2602,9 @@ impl<S: StorageManager + Send + Sync + 'static, N: NetworkManager + Send + Sync
 
     /// Get the number of filters that have been received.
     pub fn get_received_filter_count(&self) -> u32 {
-        if let Ok(heights) = self.received_filter_heights.lock() {
-            heights.len() as u32
-        } else {
-            0
+        match self.received_filter_heights.try_lock() {
+            Ok(heights) => heights.len() as u32,
+            Err(_) => 0,
         }
     }
 
@@ -2853,9 +2847,8 @@ impl<S: StorageManager + Send + Sync + 'static, N: NetworkManager + Send + Sync
             stats_lock.filter_sync_start_time = Some(std::time::Instant::now());
             stats_lock.last_filter_received_time = None;
             // Clear the received heights tracking for a fresh start
-            if let Ok(mut heights) = stats_lock.received_filter_heights.lock() {
-                heights.clear();
-            }
+            let mut heights = stats_lock.received_filter_heights.lock().await;
+            heights.clear();
             tracing::info!(
                 "📊 Started new filter sync tracking: {} filters requested",
                 total_filters_requested
@@ -2895,14 +2888,13 @@ impl<S: StorageManager + Send + Sync + 'static, N: NetworkManager + Send + Sync
             drop(stats_lock); // Release the stats lock before acquiring the mutex
 
             // Now lock the heights and insert
-            if let Ok(mut heights) = received_filter_heights.lock() {
-                heights.insert(height);
-                tracing::trace!(
-                    "📊 Recorded filter received at height {} for block {}",
-                    height,
-                    block_hash
-                );
-            };
+            let mut heights = received_filter_heights.lock().await;
+            heights.insert(height);
+            tracing::trace!(
+                "📊 Recorded filter received at height {} for block {}",
+                height,
+                block_hash
+            );
         } else {
             tracing::warn!("Could not find height for filter block hash {}", block_hash);
         }
@@ -3030,7 +3022,7 @@ impl<S: StorageManager + Send + Sync + 'static, N: NetworkManager + Send + Sync
 
     /// Record receipt of a filter at a specific height.
     pub fn record_filter_received(&mut self, height: u32) {
-        if let Ok(mut heights) = self.received_filter_heights.lock() {
+        if let Ok(mut heights) = self.received_filter_heights.try_lock() {
             heights.insert(height);
             tracing::trace!("📊 Recorded filter received at height {}", height);
         }
@@ -3040,9 +3032,9 @@ impl<S: StorageManager + Send + Sync + 'static, N: NetworkManager + Send + Sync
     pub fn find_missing_ranges(&self) -> Vec<(u32, u32)> {
         let mut missing_ranges = Vec::new();
 
-        let heights = match self.received_filter_heights.lock() {
+        let heights = match self.received_filter_heights.try_lock() {
             Ok(heights) => heights.clone(),
-            Err(_) => return missing_ranges, // Return empty if lock fails
+            Err(_) => return missing_ranges,
         };
 
         // For each requested range
@@ -3077,9 +3069,9 @@ impl<S: StorageManager + Send + Sync + 'static, N: NetworkManager + Send + Sync
         let now = std::time::Instant::now();
         let mut timed_out = Vec::new();
 
-        let heights = match self.received_filter_heights.lock() {
+        let heights = match self.received_filter_heights.try_lock() {
             Ok(heights) => heights.clone(),
-            Err(_) => return timed_out, // Return empty if lock fails
+            Err(_) => return timed_out,
         };
 
         for ((start, end), request_time) in &self.requested_filter_ranges {
@@ -3104,9 +3096,9 @@ impl<S: StorageManager + Send + Sync + 'static, N: NetworkManager + Send + Sync
 
     /// Check if a filter range is complete (all heights received).
     pub fn is_range_complete(&self, start_height: u32, end_height: u32) -> bool {
-        let heights = match self.received_filter_heights.lock() {
+        let heights = match self.received_filter_heights.try_lock() {
             Ok(heights) => heights,
-            Err(_) => return false, // Return false if lock fails
+            Err(_) => return false,
         };
 
         for height in start_height..=end_height {
@@ -3453,7 +3445,7 @@ impl<S: StorageManager + Send + Sync + 'static, N: NetworkManager + Send + Sync
     /// Reset filter range tracking (useful for testing or restart scenarios).
     pub fn reset_filter_tracking(&mut self) {
         self.requested_filter_ranges.clear();
-        if let Ok(mut heights) = self.received_filter_heights.lock() {
+        if let Ok(mut heights) = self.received_filter_heights.try_lock() {
             heights.clear();
         }
         self.filter_retry_counts.clear();

dash-spv/src/sync/sequential/mod.rs

@@ -85,7 +85,7 @@ impl<
     /// Create a new sequential sync manager
     pub fn new(
         config: &ClientConfig,
-        received_filter_heights: std::sync::Arc<std::sync::Mutex<std::collections::HashSet<u32>>>,
+        received_filter_heights: std::sync::Arc<tokio::sync::Mutex<std::collections::HashSet<u32>>>,
         wallet: std::sync::Arc<tokio::sync::RwLock<W>>,
     ) -> SyncResult<Self> {
         // Create reorg config with sensible defaults

dash-spv/src/types.rs

@@ -550,7 +550,7 @@ pub struct SpvStats {
 
     /// Received filter heights for gap tracking (shared with FilterSyncManager).
     #[serde(skip)]
-    pub received_filter_heights: std::sync::Arc<std::sync::Mutex<std::collections::HashSet<u32>>>,
+    pub received_filter_heights: std::sync::Arc<tokio::sync::Mutex<std::collections::HashSet<u32>>>,
 
     /// Number of filter requests currently active.
     pub active_filter_requests: u32,
@@ -587,7 +587,7 @@ impl Default for SpvStats {
             filters_received: 0,
             filter_sync_start_time: None,
             last_filter_received_time: None,
-            received_filter_heights: std::sync::Arc::new(std::sync::Mutex::new(
+            received_filter_heights: std::sync::Arc::new(tokio::sync::Mutex::new(
                 std::collections::HashSet::new(),
             )),
             active_filter_requests: 0,

dash-spv/tests/block_download_test.rs

@@ -8,7 +8,8 @@
 //! Tests for block downloading on filter match functionality.
 
 use std::collections::HashSet;
-use std::sync::{Arc, Mutex};
+use std::sync::Arc;
+use tokio::sync::Mutex;
 use tokio::sync::RwLock;
 
 use dashcore::{

dash-spv/tests/cfheader_gap_test.rs

@@ -6,7 +6,8 @@
 //! Tests for CFHeader gap detection and auto-restart functionality.
 
 use std::collections::HashSet;
-use std::sync::{Arc, Mutex};
+use std::sync::Arc;
+use tokio::sync::Mutex;
 
 use dash_spv::{
     client::ClientConfig,

dash-spv/tests/edge_case_filter_sync_test.rs

@@ -8,7 +8,8 @@
 //! Tests for edge case handling in filter header sync, particularly at the tip.
 
 use std::collections::HashSet;
-use std::sync::{Arc, Mutex};
+use std::sync::Arc;
+use tokio::sync::Mutex;
 
 use dash_spv::{
     client::ClientConfig,
@@ -52,8 +53,8 @@ impl MockNetworkManager {
         }
     }
 
-    fn get_sent_messages(&self) -> Vec<NetworkMessage> {
-        self.sent_messages.lock().unwrap().clone()
+    async fn get_sent_messages(&self) -> Vec<NetworkMessage> {
+        self.sent_messages.lock().await.clone()
     }
 }
 
@@ -72,7 +73,7 @@ impl NetworkManager for MockNetworkManager {
     }
 
     async fn send_message(&mut self, message: NetworkMessage) -> NetworkResult<()> {
-        self.sent_messages.lock().unwrap().push(message);
+        self.sent_messages.lock().await.push(message);
         Ok(())
     }
 
@@ -181,7 +182,7 @@ async fn test_filter_sync_at_tip_edge_case() {
     assert!(!result.unwrap(), "Should not start sync when already at tip");
 
     // Verify no messages were sent
-    let sent_messages = network.get_sent_messages();
+    let sent_messages = network.get_sent_messages().await;
     assert_eq!(sent_messages.len(), 0, "Should not send any messages when at tip");
 }
 
@@ -285,7 +286,7 @@ async fn test_no_invalid_getcfheaders_at_tip() {
     assert!(result.unwrap(), "Should start sync when behind by 1 block");
 
     // Check the sent message
-    let sent_messages = network.get_sent_messages();
+    let sent_messages = network.get_sent_messages().await;
     assert_eq!(sent_messages.len(), 1, "Should send exactly one message");
 
     match &sent_messages[0] {

dash-spv/tests/filter_header_verification_test.rs

@@ -32,7 +32,8 @@ use dashcore::{
 };
 use dashcore_hashes::{sha256d, Hash};
 use std::collections::HashSet;
-use std::sync::{Arc, Mutex};
+use std::sync::Arc;
+use tokio::sync::Mutex;
 
 /// Mock network manager for testing filter sync
 #[derive(Debug)]

dash-spv/tests/simple_gap_test.rs

@@ -1,7 +1,8 @@
 //! Basic test for CFHeader gap detection functionality.
 
 use std::collections::HashSet;
-use std::sync::{Arc, Mutex};
+use std::sync::Arc;
+use tokio::sync::Mutex;
 
 use dash_spv::{
     client::ClientConfig,