Squashed commit of the following:

commit 2e7ea89
Author: xdustinface <[email protected]>
Date:   Sun Nov 16 23:51:16 2025 +1000

    fix: Use non-blocking `TcpStream` in `dash-spv::network::TcpConnection`

Author: xdustinface

dash-spv-ffi/Cargo.toml

@@ -17,6 +17,7 @@ dashcore = { path = "../dash", package = "dashcore" }
 libc = "0.2"
 once_cell = "1.19"
 tokio = { version = "1", features = ["full"] }
+tokio-util = "0.7"
 serde = { version = "1.0", features = ["derive"] }
 serde_json = "1.0"
 log = "0.4"

dash-spv-ffi/src/client.rs

@@ -17,11 +17,12 @@ use std::collections::HashMap;
 use std::ffi::{CStr, CString};
 use std::os::raw::{c_char, c_void};
 use std::str::FromStr;
-use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
+use std::sync::atomic::{AtomicU64, Ordering};
 use std::sync::{Arc, Mutex};
 use std::time::Duration;
 use tokio::runtime::Runtime;
 use tokio::sync::mpsc::{error::TryRecvError, UnboundedReceiver};
+use tokio_util::sync::CancellationToken;
 
 /// Global callback registry for thread-safe callback management
 static CALLBACK_REGISTRY: Lazy<Arc<Mutex<CallbackRegistry>>> =
@@ -104,12 +105,6 @@ struct SyncCallbackData {
     _marker: std::marker::PhantomData<()>,
 }
 
-async fn wait_for_shutdown_signal(signal: Arc<AtomicBool>) {
-    while !signal.load(Ordering::Relaxed) {
-        tokio::time::sleep(Duration::from_millis(50)).await;
-    }
-}
-
 /// FFIDashSpvClient structure
 type InnerClient = DashSpvClient<
     key_wallet_manager::wallet_manager::WalletManager<
@@ -126,7 +121,7 @@ pub struct FFIDashSpvClient {
     event_callbacks: Arc<Mutex<FFIEventCallbacks>>,
     active_threads: Arc<Mutex<Vec<std::thread::JoinHandle<()>>>>,
     sync_callbacks: Arc<Mutex<Option<SyncCallbackData>>>,
-    shutdown_signal: Arc<AtomicBool>,
+    shutdown_token: CancellationToken,
     // Stored event receiver for pull-based draining (no background thread by default)
     event_rx: Arc<Mutex<Option<UnboundedReceiver<dash_spv::types::SpvEvent>>>>,
 }
@@ -197,7 +192,7 @@ pub unsafe extern "C" fn dash_spv_ffi_client_new(
                 event_callbacks: Arc::new(Mutex::new(FFIEventCallbacks::default())),
                 active_threads: Arc::new(Mutex::new(Vec::new())),
                 sync_callbacks: Arc::new(Mutex::new(None)),
-                shutdown_signal: Arc::new(AtomicBool::new(false)),
+                shutdown_token: CancellationToken::new(),
                 event_rx: Arc::new(Mutex::new(None)),
             };
             Box::into_raw(Box::new(ffi_client))
@@ -378,8 +373,8 @@ pub unsafe extern "C" fn dash_spv_ffi_client_drain_events(client: *mut FFIDashSp
     FFIErrorCode::Success as i32
 }
 
-fn stop_client_internal(client: &FFIDashSpvClient) -> Result<(), dash_spv::SpvError> {
-    client.shutdown_signal.store(true, Ordering::Relaxed);
+fn stop_client_internal(client: &mut FFIDashSpvClient) -> Result<(), dash_spv::SpvError> {
+    client.shutdown_token.cancel();
 
     // Ensure callbacks are cleared so no further progress/completion notifications fire.
     {
@@ -411,7 +406,7 @@ fn stop_client_internal(client: &FFIDashSpvClient) -> Result<(), dash_spv::SpvEr
         res
     });
 
-    client.shutdown_signal.store(false, Ordering::Relaxed);
+    client.shutdown_token = CancellationToken::new();
 
     result
 }
@@ -525,7 +520,7 @@ pub unsafe extern "C" fn dash_spv_ffi_client_start(client: *mut FFIDashSpvClient
 pub unsafe extern "C" fn dash_spv_ffi_client_stop(client: *mut FFIDashSpvClient) -> i32 {
     null_check!(client);
 
-    let client = &(*client);
+    let client = &mut (*client);
     match stop_client_internal(client) {
         Ok(()) => FFIErrorCode::Success as i32,
         Err(e) => {
@@ -785,7 +780,6 @@ pub unsafe extern "C" fn dash_spv_ffi_client_sync_to_tip_with_progress(
     let inner = client.inner.clone();
     let runtime = client.runtime.clone();
     let sync_callbacks = client.sync_callbacks.clone();
-    let shutdown_signal = client.shutdown_signal.clone();
 
     // Take progress receiver from client
     let progress_receiver = {
@@ -797,7 +791,7 @@ pub unsafe extern "C" fn dash_spv_ffi_client_sync_to_tip_with_progress(
     if let Some(mut receiver) = progress_receiver {
         let runtime_handle = runtime.handle().clone();
         let sync_callbacks_clone = sync_callbacks.clone();
-        let shutdown_signal_clone = shutdown_signal.clone();
+        let shutdown_token_monitor = client.shutdown_token.clone();
 
         let handle = std::thread::spawn(move || {
             runtime_handle.block_on(async move {
@@ -859,7 +853,7 @@ pub unsafe extern "C" fn dash_spv_ffi_client_sync_to_tip_with_progress(
                                 None => break,
                             }
                         }
-                        _ = wait_for_shutdown_signal(shutdown_signal_clone.clone()) => {
+                        _ = shutdown_token_monitor.cancelled() => {
                             break;
                         }
                     }
@@ -874,15 +868,12 @@ pub unsafe extern "C" fn dash_spv_ffi_client_sync_to_tip_with_progress(
     // Spawn sync task in a separate thread with safe callback access
     let runtime_handle = runtime.handle().clone();
     let sync_callbacks_clone = sync_callbacks.clone();
-    let shutdown_signal_for_thread = shutdown_signal.clone();
-    let stop_triggered_for_thread = Arc::new(AtomicBool::new(false));
+    let shutdown_token_sync = client.shutdown_token.clone();
     let sync_handle = std::thread::spawn(move || {
-        let stop_triggered_for_callback = stop_triggered_for_thread.clone();
+        let shutdown_token_callback = shutdown_token_sync.clone();
         // Run monitoring loop
         let monitor_result = runtime_handle.block_on({
             let inner = inner.clone();
-            let shutdown_signal_for_thread = shutdown_signal_for_thread.clone();
-            let stop_triggered_for_thread = stop_triggered_for_callback.clone();
             async move {
                 let mut spv_client = {
                     let mut guard = inner.lock().unwrap();
@@ -903,8 +894,7 @@ pub unsafe extern "C" fn dash_spv_ffi_client_sync_to_tip_with_progress(
                         Ok(inner) => inner,
                         Err(_) => Ok(()),
                     },
-                    _ = wait_for_shutdown_signal(shutdown_signal_for_thread.clone()) => {
-                        stop_triggered_for_thread.store(true, Ordering::Relaxed);
+                    _ = shutdown_token_sync.cancelled() => {
                         abort_handle.abort();
                         match monitor_future.as_mut().await {
                             Ok(inner) => inner,
@@ -930,7 +920,7 @@ pub unsafe extern "C" fn dash_spv_ffi_client_sync_to_tip_with_progress(
                     ..
                 }) = registry.unregister(callback_data.callback_id)
                 {
-                    if stop_triggered_for_callback.load(Ordering::Relaxed) {
+                    if shutdown_token_callback.is_cancelled() {
                         let msg = CString::new("Sync stopped by request").unwrap_or_else(|_| {
                             CString::new("Sync stopped").expect("hardcoded string is safe")
                         });
@@ -984,7 +974,7 @@ pub unsafe extern "C" fn dash_spv_ffi_client_sync_to_tip_with_progress(
 pub unsafe extern "C" fn dash_spv_ffi_client_cancel_sync(client: *mut FFIDashSpvClient) -> i32 {
     null_check!(client);
 
-    let client = &(*client);
+    let client = &mut (*client);
 
     match stop_client_internal(client) {
         Ok(()) => FFIErrorCode::Success as i32,
@@ -1318,8 +1308,8 @@ pub unsafe extern "C" fn dash_spv_ffi_client_destroy(client: *mut FFIDashSpvClie
     if !client.is_null() {
         let client = Box::from_raw(client);
 
-        // Set shutdown signal to stop all threads
-        client.shutdown_signal.store(true, Ordering::Relaxed);
+        // Cancel shutdown token to stop all threads
+        client.shutdown_token.cancel();
 
         // Clean up any registered callbacks
         if let Some(ref callback_data) = *client.sync_callbacks.lock().unwrap() {

dash-spv/ARCHITECTURE.md

@@ -680,7 +680,7 @@ The network module handles all P2P communication with the Dash network.
 - Graceful shutdown
 
 **Complex Types Used**:
-- `HashMap<PeerId, Arc<TcpConnection>>` - **JUSTIFIED**: Efficient peer lookup
+- `HashMap<PeerId, Arc<Peer>>` - **JUSTIFIED**: Efficient peer lookup
 - Multiple tokio::sync primitives - **JUSTIFIED**: Complex concurrent operations
 
 **Critical Issues**:

dash-spv/CLAUDE.md

@@ -116,7 +116,7 @@ TCP-based networking with proper Dash protocol implementation:
 - **DNS-first peer discovery**: Automatically uses DNS seeds (`dnsseed.dash.org`, `testnet-seed.dashdot.io`) when no explicit peers are configured
 - **Immediate startup**: No delay for initial peer discovery (10-second delay only for subsequent searches)
 - **Exclusive mode**: When explicit peers are provided, uses only those peers (no DNS discovery)
-- Connection management via `TcpConnection`
+- Connection management via `Peer`
 - Handshake handling via `HandshakeManager`
 - Message routing via `MessageHandler`
 - Peer support via `PeerNetworkManager`

dash-spv/Cargo.toml

@@ -23,6 +23,7 @@ clap = { version = "4.0", features = ["derive"] }
 
 # Async runtime
 tokio = { version = "1.0", features = ["full"] }
+tokio-util = "0.7"
 async-trait = "0.1"
 
 # Error handling

dash-spv/src/client/config.rs

@@ -56,9 +56,6 @@ pub struct ClientConfig {
     /// Sync timeout.
     pub sync_timeout: Duration,
 
-    /// Read timeout for TCP socket operations.
-    pub read_timeout: Duration,
-
     /// Whether to enable filter syncing.
     pub enable_filters: bool,
 
@@ -206,7 +203,6 @@ impl Default for ClientConfig {
             connection_timeout: Duration::from_secs(30),
             message_timeout: Duration::from_secs(60),
             sync_timeout: Duration::from_secs(300),
-            read_timeout: Duration::from_millis(100),
             enable_filters: true,
             enable_masternodes: true,
             max_peers: 8,
@@ -326,12 +322,6 @@ impl ClientConfig {
         self
     }
 
-    /// Set read timeout for TCP socket operations.
-    pub fn with_read_timeout(mut self, timeout: Duration) -> Self {
-        self.read_timeout = timeout;
-        self
-    }
-
     /// Set log level.
     pub fn with_log_level(mut self, level: &str) -> Self {
         self.log_level = level.to_string();

dash-spv/src/client/config_test.rs

@@ -22,7 +22,6 @@ mod tests {
         assert_eq!(config.connection_timeout, Duration::from_secs(30));
         assert_eq!(config.message_timeout, Duration::from_secs(60));
         assert_eq!(config.sync_timeout, Duration::from_secs(300));
-        assert_eq!(config.read_timeout, Duration::from_millis(100));
         assert!(config.enable_filters);
         assert!(config.enable_masternodes);
         assert_eq!(config.max_peers, 8);
@@ -65,7 +64,6 @@ mod tests {
             .with_storage_path(path.clone())
             .with_validation_mode(ValidationMode::Basic)
             .with_connection_timeout(Duration::from_secs(10))
-            .with_read_timeout(Duration::from_secs(5))
             .with_log_level("debug")
             .with_max_concurrent_filter_requests(32)
             .with_filter_flow_control(false)
@@ -80,7 +78,6 @@ mod tests {
         assert!(config.enable_persistence);
         assert_eq!(config.validation_mode, ValidationMode::Basic);
         assert_eq!(config.connection_timeout, Duration::from_secs(10));
-        assert_eq!(config.read_timeout, Duration::from_secs(5));
         assert_eq!(config.log_level, "debug");
         assert_eq!(config.max_concurrent_filter_requests, 32);
         assert!(!config.enable_filter_flow_control);

dash-spv/src/network/handshake.rs

@@ -12,7 +12,7 @@ use dashcore::Network;
 
 use crate::client::config::MempoolStrategy;
 use crate::error::{NetworkError, NetworkResult};
-use crate::network::connection::TcpConnection;
+use crate::network::peer::Peer;
 
 /// Handshake state.
 #[derive(Debug, Clone, PartialEq, Eq)]
@@ -65,7 +65,7 @@ impl HandshakeManager {
     }
 
     /// Perform the handshake with a peer.
-    pub async fn perform_handshake(&mut self, connection: &mut TcpConnection) -> NetworkResult<()> {
+    pub async fn perform_handshake(&mut self, connection: &mut Peer) -> NetworkResult<()> {
         use tokio::time::{timeout, Duration};
 
         // Send version message
@@ -145,7 +145,7 @@ impl HandshakeManager {
     /// Handle a handshake message.
     async fn handle_handshake_message(
         &mut self,
-        connection: &mut TcpConnection,
+        connection: &mut Peer,
         message: NetworkMessage,
     ) -> NetworkResult<Option<HandshakeState>> {
         match message {
@@ -238,7 +238,7 @@ impl HandshakeManager {
     }
 
     /// Send version message.
-    async fn send_version(&mut self, connection: &mut TcpConnection) -> NetworkResult<()> {
+    async fn send_version(&mut self, connection: &mut Peer) -> NetworkResult<()> {
         let version_message = self.build_version_message(connection.peer_info().address)?;
         connection.send_message(NetworkMessage::Version(version_message)).await?;
         tracing::debug!("Sent version message");
@@ -309,7 +309,7 @@ impl HandshakeManager {
     }
 
     /// Negotiate headers2 support with the peer after handshake completion.
-    async fn negotiate_headers2(&self, connection: &mut TcpConnection) -> NetworkResult<()> {
+    async fn negotiate_headers2(&self, connection: &mut Peer) -> NetworkResult<()> {
         // Headers2 is currently disabled due to protocol compatibility issues
         // Always send SendHeaders regardless of peer support
         tracing::info!("Headers2 is disabled - sending SendHeaders only");