feat: Introduce DashSpvClientInterface

Author: xdustinface

dash-spv-ffi/src/client.rs

@@ -886,23 +886,27 @@ pub unsafe extern "C" fn dash_spv_ffi_client_sync_to_tip_with_progress(
                         }
                     }
                 };
+                let (_command_sender, command_receiver) = tokio::sync::mpsc::unbounded_channel();
+                let run_token = shutdown_token_sync.clone();
                 let (abort_handle, abort_registration) = AbortHandle::new_pair();
-                let mut monitor_future =
-                    Box::pin(Abortable::new(spv_client.monitor_network(), abort_registration));
+                let mut run_future = Box::pin(Abortable::new(
+                    spv_client.run(command_receiver, run_token),
+                    abort_registration,
+                ));
                 let result = tokio::select! {
-                    res = &mut monitor_future => match res {
+                    res = &mut run_future => match res {
                         Ok(inner) => inner,
                         Err(_) => Ok(()),
                     },
                     _ = shutdown_token_sync.cancelled() => {
                         abort_handle.abort();
-                        match monitor_future.as_mut().await {
+                        match run_future.as_mut().await {
                             Ok(inner) => inner,
                             Err(_) => Ok(()),
                         }
                     }
                 };
-                drop(monitor_future);
+                drop(run_future);
                 let mut guard = inner.lock().unwrap();
                 *guard = Some(spv_client);
                 result

dash-spv-ffi/src/error.rs

@@ -52,6 +52,7 @@ pub extern "C" fn dash_spv_ffi_clear_error() {
 impl From<SpvError> for FFIErrorCode {
     fn from(err: SpvError) -> Self {
         match err {
+            SpvError::ChannelFailure(_, _) => FFIErrorCode::RuntimeError,
             SpvError::Network(_) => FFIErrorCode::NetworkError,
             SpvError::Storage(_) => FFIErrorCode::StorageError,
             SpvError::Validation(_) => FFIErrorCode::ValidationError,
@@ -60,6 +61,7 @@ impl From<SpvError> for FFIErrorCode {
             SpvError::Config(_) => FFIErrorCode::ConfigError,
             SpvError::Parse(_) => FFIErrorCode::ValidationError,
             SpvError::Wallet(_) => FFIErrorCode::WalletError,
+            SpvError::QuorumLookupError(_) => FFIErrorCode::ValidationError,
             SpvError::General(_) => FFIErrorCode::Unknown,
         }
     }

dash-spv/examples/filter_sync.rs

@@ -8,8 +8,8 @@ use key_wallet::wallet::managed_wallet_info::ManagedWalletInfo;
 use key_wallet_manager::wallet_manager::WalletManager;
 use std::str::FromStr;
 use std::sync::Arc;
-use tokio::signal;
 use tokio::sync::RwLock;
+use tokio_util::sync::CancellationToken;
 
 #[tokio::main]
 async fn main() -> Result<(), Box<dyn std::error::Error>> {
@@ -43,29 +43,12 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     println!("Watching address: {:?}", watch_address);
 
     // Full sync including filters
-    let progress = client.sync_to_tip().await?;
+    client.sync_to_tip().await?;
 
-    tokio::select! {
-        result = client.monitor_network() => {
-            println!("monitor_network result {:?}", result);
-        },
-        _ = signal::ctrl_c() => {
-            println!("monitor_network canceled");
-        }
-    }
+    let (_command_sender, command_receiver) = tokio::sync::mpsc::unbounded_channel();
+    let shutdown_token = CancellationToken::new();
 
-    println!("Headers synced: {}", progress.header_height);
-    println!("Filter headers synced: {}", progress.filter_header_height);
-
-    // Get statistics
-    let stats = client.stats().await?;
-    println!("Filter headers downloaded: {}", stats.filter_headers_downloaded);
-    println!("Filters downloaded: {}", stats.filters_downloaded);
-    println!("Filter matches found: {}", stats.filters_matched);
-    println!("Blocks requested: {}", stats.blocks_requested);
-
-    // Stop the client
-    client.stop().await?;
+    client.run_until_shutdown(command_receiver, shutdown_token).await?;
 
     println!("Done!");
     Ok(())

dash-spv/examples/simple_sync.rs

@@ -7,8 +7,8 @@ use key_wallet::wallet::managed_wallet_info::ManagedWalletInfo;
 
 use key_wallet_manager::wallet_manager::WalletManager;
 use std::sync::Arc;
-use tokio::signal;
 use tokio::sync::RwLock;
+use tokio_util::sync::CancellationToken;
 
 #[tokio::main]
 async fn main() -> Result<(), Box<dyn std::error::Error>> {
@@ -48,17 +48,10 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     println!("Headers downloaded: {}", stats.headers_downloaded);
     println!("Bytes received: {}", stats.bytes_received);
 
-    tokio::select! {
-        result = client.monitor_network() => {
-            println!("monitor_network result {:?}", result);
-        },
-        _ = signal::ctrl_c() => {
-            println!("monitor_network canceled");
-        }
-    }
-
-    // Stop the client
-    client.stop().await?;
+    let (_command_sender, command_receiver) = tokio::sync::mpsc::unbounded_channel();
+    let shutdown_token = CancellationToken::new();
+
+    client.run_until_shutdown(command_receiver, shutdown_token).await?;
 
     println!("Done!");
     Ok(())

dash-spv/examples/spv_with_wallet.rs

@@ -8,8 +8,8 @@ use dash_spv::{ClientConfig, DashSpvClient};
 use key_wallet::wallet::managed_wallet_info::ManagedWalletInfo;
 use key_wallet_manager::wallet_manager::WalletManager;
 use std::sync::Arc;
-use tokio::signal;
 use tokio::sync::RwLock;
+use tokio_util::sync::CancellationToken;
 
 #[tokio::main]
 async fn main() -> Result<(), Box<dyn std::error::Error>> {
@@ -48,18 +48,11 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     // - Mempool transactions via process_mempool_transaction()
     // - Reorgs via handle_reorg()
     // - Compact filter checks via check_compact_filter()
-    tokio::select! {
-        result = client.monitor_network() => {
-            println!("monitor_network result {:?}", result);
-        },
-        _ = signal::ctrl_c() => {
-            println!("monitor_network canceled");
-        }
-    }
 
-    // Stop the client
-    println!("Stopping SPV client...");
-    client.stop().await?;
+    let (_command_sender, command_receiver) = tokio::sync::mpsc::unbounded_channel();
+    let shutdown_token = CancellationToken::new();
+
+    client.run_until_shutdown(command_receiver, shutdown_token).await?;
 
     println!("Done!");
     Ok(())

dash-spv/src/client/interface.rs

@@ -0,0 +1,79 @@
+use crate::error::SpvError;
+use dashcore::sml::llmq_type::LLMQType;
+use dashcore::sml::quorum_entry::qualified_quorum_entry::QualifiedQuorumEntry;
+use dashcore::QuorumHash;
+use std::fmt::Display;
+use tokio::sync::{mpsc, oneshot};
+
+pub type Result<T> = std::result::Result<T, SpvError>;
+
+pub type GetQuorumByHeightResult = Result<QualifiedQuorumEntry>;
+
+async fn receive<Type>(context: String, receiver: oneshot::Receiver<Type>) -> Result<Type> {
+    receiver.await.map_err(|error| SpvError::ChannelFailure(context, error.to_string()))
+}
+
+pub enum DashSpvClientCommand {
+    GetQuorumByHeight {
+        height: u32,
+        quorum_type: LLMQType,
+        quorum_hash: QuorumHash,
+        sender: oneshot::Sender<GetQuorumByHeightResult>,
+    },
+}
+
+impl DashSpvClientCommand {
+    pub async fn send(
+        self,
+        context: String,
+        sender: mpsc::UnboundedSender<DashSpvClientCommand>,
+    ) -> Result<()> {
+        sender.send(self).map_err(|error| SpvError::ChannelFailure(context, error.to_string()))?;
+        Ok(())
+    }
+}
+
+impl Display for DashSpvClientCommand {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        let str = match self {
+            DashSpvClientCommand::GetQuorumByHeight {
+                height,
+                quorum_type,
+                quorum_hash,
+                sender: _,
+            } => format!("GetQuorumByHeight({height}, {quorum_type}, {quorum_hash})"),
+        };
+        write!(f, "{}", str)
+    }
+}
+
+#[derive(Clone)]
+pub struct DashSpvClientInterface {
+    pub command_sender: mpsc::UnboundedSender<DashSpvClientCommand>,
+}
+
+impl DashSpvClientInterface {
+    pub fn new(command_sender: mpsc::UnboundedSender<DashSpvClientCommand>) -> Self {
+        Self {
+            command_sender,
+        }
+    }
+
+    pub async fn get_quorum_by_height(
+        &self,
+        height: u32,
+        quorum_type: LLMQType,
+        quorum_hash: QuorumHash,
+    ) -> GetQuorumByHeightResult {
+        let (sender, receiver) = oneshot::channel();
+        let command = DashSpvClientCommand::GetQuorumByHeight {
+            height,
+            quorum_type,
+            quorum_hash,
+            sender,
+        };
+        let context = command.to_string();
+        command.send(context.clone(), self.command_sender.clone()).await?;
+        receive(context, receiver).await?
+    }
+}

dash-spv/src/client/mod.rs

@@ -37,6 +37,7 @@
 pub mod block_processor;
 pub mod config;
 pub mod filter_sync;
+pub mod interface;
 pub mod message_handler;
 pub mod status_display;
 

dash-spv/src/client/queries.rs

@@ -10,9 +10,11 @@ use crate::error::{Result, SpvError};
 use crate::network::NetworkManager;
 use crate::storage::StorageManager;
 use crate::types::AddressBalance;
+use dashcore::sml::llmq_type::LLMQType;
 use dashcore::sml::masternode_list::MasternodeList;
 use dashcore::sml::masternode_list_engine::MasternodeListEngine;
 use dashcore::sml::quorum_entry::qualified_quorum_entry::QualifiedQuorumEntry;
+use dashcore::QuorumHash;
 use key_wallet_manager::wallet_interface::WalletInterface;
 
 use super::DashSpvClient;
@@ -73,45 +75,32 @@ impl<
     pub fn get_quorum_at_height(
         &self,
         height: u32,
-        quorum_type: u8,
-        quorum_hash: &[u8; 32],
-    ) -> Option<&QualifiedQuorumEntry> {
-        use dashcore::sml::llmq_type::LLMQType;
-        use dashcore::QuorumHash;
-        use dashcore_hashes::Hash;
-
-        let llmq_type: LLMQType = LLMQType::from(quorum_type);
-        if llmq_type == LLMQType::LlmqtypeUnknown {
-            tracing::warn!("Invalid quorum type {} requested at height {}", quorum_type, height);
-            return None;
-        };
-
-        let qhash = QuorumHash::from_byte_array(*quorum_hash);
-
+        quorum_type: LLMQType,
+        quorum_hash: QuorumHash,
+    ) -> Result<QualifiedQuorumEntry> {
         // First check if we have the masternode list at this height
         match self.get_masternode_list_at_height(height) {
             Some(ml) => {
                 // We have the masternode list, now look for the quorum
-                match ml.quorums.get(&llmq_type) {
-                    Some(quorums) => match quorums.get(&qhash) {
+                match ml.quorums.get(&quorum_type) {
+                    Some(quorums) => match quorums.get(&quorum_hash) {
                         Some(quorum) => {
                             tracing::debug!(
                                 "Found quorum type {} at height {} with hash {}",
                                 quorum_type,
                                 height,
                                 hex::encode(quorum_hash)
                             );
-                            Some(quorum)
+                            Ok(quorum.clone())
                         }
                         None => {
-                            tracing::warn!(
-                                "Quorum not found: type {} at height {} with hash {} (masternode list exists with {} quorums of this type)",
-                                quorum_type,
-                                height,
-                                hex::encode(quorum_hash),
-                                quorums.len()
-                            );
-                            None
+                            let message = format!("Quorum not found: type {} at height {} with hash {} (masternode list exists with {} quorums of this type)",
+                                                quorum_type,
+                                                height,
+                                                hex::encode(quorum_hash),
+                                                quorums.len());
+                            tracing::warn!(message);
+                            Err(SpvError::QuorumLookupError(message))
                         }
                     },
                     None => {
@@ -120,7 +109,10 @@ impl<
                             quorum_type,
                             height
                         );
-                        None
+                        Err(SpvError::QuorumLookupError(format!(
+                            "No quorums of type {} found at height {}",
+                            quorum_type, height
+                        )))
                     }
                 }
             }
@@ -129,7 +121,10 @@ impl<
                     "No masternode list found at height {} - cannot retrieve quorum",
                     height
                 );
-                None
+                Err(SpvError::QuorumLookupError(format!(
+                    "No masternode list found at height {}",
+                    height
+                )))
             }
         }
     }