refactor: migrate received_filter_heights to tokio::Mutex and add find_available_header_at_or_before helper (#141)

* refactor: streamline header retrieval in FilterSyncManager

- Introduced `find_available_header_at_or_before` method to encapsulate the logic for scanning backward to find the nearest available block header in storage.
- Replaced repetitive header scanning logic in multiple locations with calls to the new method, improving code clarity and maintainability.
- Enhanced error handling and logging during header retrieval to provide better insights into the syncing process.

* 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.

* refactor: optimize sync progress calculation in StatusDisplay

- Improved the sync progress method by cloning the inner heights handle and copying necessary counters without holding the RwLock, enhancing performance.
- Updated the calculation of the last synced filter height to avoid holding the RwLock guard, ensuring better concurrency and responsiveness in the status display.

* refactor: introduce SharedFilterHeights type for improved readability

- Created a new type alias `SharedFilterHeights` for the mutex-protected set of filter heights, enhancing code clarity and reducing redundancy.
- Updated references in `SpvStats`, `FilterSyncManager`, and `SequentialSyncManager` to use the new type alias, streamlining the codebase.

* fix: improve error handling in filter header sync

- Updated the error handling in the filter header synchronization process to return a more informative `SyncError` when no available headers are found between specified heights.
- This change enhances the clarity of error reporting, aiding in debugging and improving the overall robustness of the synchronization logic.

* refactor: update processing_thread_requests to use tokio::sync::Mutex

- Changed the locking mechanism for `processing_thread_requests` from `std::sync::Mutex` to `tokio::sync::Mutex` to enhance async compatibility.
- Updated the locking logic in the `spawn_filter_processor` and related methods to use async await syntax, improving performance and responsiveness in the filter processing workflow.

* refactor: improve filter sync locking mechanism

- Updated the locking logic in the filter synchronization process to clone the `received_filter_heights` before awaiting the mutex, ensuring the `RwLock` is released promptly.
- This change enhances concurrency and responsiveness during filter sync operations, aligning with recent async improvements in the codebase.

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

@@ -73,13 +73,16 @@ impl<'a, S: StorageManager + Send + Sync + 'static> StatusDisplay<'a, S> {
     /// Get current sync progress.
     pub async fn sync_progress(&self) -> Result<SyncProgress> {
         let state = self.state.read().await;
-        let stats = self.stats.read().await;
+        // Clone the inner heights handle and copy needed counters without awaiting while holding the RwLock
+        let (filters_received, received_heights) = {
+            let stats = self.stats.read().await;
+            (stats.filters_received, std::sync::Arc::clone(&stats.received_filter_heights))
+        };
 
-        // Calculate last synced filter height from received filter heights
-        let last_synced_filter_height = if let Ok(heights) = stats.received_filter_heights.lock() {
+        // Calculate last synced filter height from received filter heights without holding the RwLock guard
+        let last_synced_filter_height = {
+            let heights = received_heights.lock().await;
             heights.iter().max().copied()
-        } else {
-            None
         };
 
         // Calculate the actual header height considering checkpoint sync
@@ -100,7 +103,7 @@ impl<'a, S: StorageManager + Send + Sync + 'static> StatusDisplay<'a, S> {
             filter_headers_synced: false, // TODO: Implement
             masternodes_synced: false,    // TODO: Implement
             filter_sync_available: false, // TODO: Get from network manager
-            filters_downloaded: stats.filters_received,
+            filters_downloaded: filters_received,
             last_synced_filter_height,
             sync_start: std::time::SystemTime::now(), // TODO: Track properly
             last_update: std::time::SystemTime::now(),

dash-spv/src/sync/filters.rs

@@ -24,7 +24,7 @@ use crate::storage::StorageManager;
 use crate::sync::{
     FilterSyncManager, HeaderSyncManagerWithReorg, MasternodeSyncManager, ReorgConfig,
 };
-use crate::types::{ChainState, SyncProgress};
+use crate::types::{ChainState, SharedFilterHeights, SyncProgress};
 use key_wallet_manager::wallet_interface::WalletInterface;
 
 use phases::{PhaseTransition, SyncPhase};
@@ -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: SharedFilterHeights,
         wallet: std::sync::Arc<tokio::sync::RwLock<W>>,
     ) -> SyncResult<Self> {
         // Create reorg config with sensible defaults

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

@@ -9,6 +9,9 @@ use dashcore::{
 };
 use serde::{Deserialize, Serialize};
 
+/// Shared, mutex-protected set of filter heights used across components.
+pub type SharedFilterHeights = std::sync::Arc<tokio::sync::Mutex<std::collections::HashSet<u32>>>;
+
 /// Unique identifier for a peer connection.
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
 pub struct PeerId(pub u64);
@@ -550,7 +553,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: SharedFilterHeights,
 
     /// Number of filter requests currently active.
     pub active_filter_requests: u32,
@@ -587,7 +590,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/src/types.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/block_download_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/cfheader_gap_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/edge_case_filter_sync_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/filter_header_verification_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,