diff --git a/Cargo.lock b/Cargo.lock index 32ca6b6a545..e1cea89363d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3295,7 +3295,6 @@ dependencies = [ "hyper", "hyper-util", "rustls", - "rustls-native-certs", "rustls-pki-types", "tokio", "tokio-rustls", @@ -5578,7 +5577,6 @@ dependencies = [ "pin-project-lite", "quinn", "rustls", - "rustls-native-certs", "rustls-pki-types", "serde", "serde_json", @@ -5877,7 +5875,6 @@ dependencies = [ "once_cell", "platform-wallet-ffi", "rand 0.8.5", - "reqwest 0.12.28", "rs-sdk-trusted-context-provider", "serde", "serde_json", @@ -5896,7 +5893,6 @@ dependencies = [ "arc-swap", "dash-context-provider", "dpp", - "futures", "hex", "lru", "reqwest 0.12.28", diff --git a/packages/rs-sdk-ffi/Cargo.toml b/packages/rs-sdk-ffi/Cargo.toml index 7ba7d20a4a0..0270b099634 100644 --- a/packages/rs-sdk-ffi/Cargo.toml +++ b/packages/rs-sdk-ffi/Cargo.toml @@ -55,8 +55,6 @@ zeroize = "1.8" # Concurrency once_cell = "1.20" -# HTTP client for diagnostics -reqwest = { version = "0.12", features = ["json", "rustls-tls-native-roots"] } [build-dependencies] cbindgen = "0.27" diff --git a/packages/rs-sdk-ffi/src/sdk.rs b/packages/rs-sdk-ffi/src/sdk.rs index bfea1fe335c..a3b3edec77f 100644 --- a/packages/rs-sdk-ffi/src/sdk.rs +++ b/packages/rs-sdk-ffi/src/sdk.rs @@ -2,7 +2,7 @@ use std::sync::{Arc, OnceLock}; use tokio::runtime::Runtime; -use tracing::{debug, error, info, warn}; +use tracing::{error, info, warn}; use dash_sdk::dpp::dashcore::Network; use dash_sdk::dpp::serialization::PlatformDeserializableWithPotentialValidationFromVersionedStructure; @@ -506,21 +506,6 @@ pub unsafe extern "C" fn dash_sdk_create_trusted(config: *const DashSDKConfig) - let runtime_clone = runtime.handle().clone(); runtime_clone.spawn(async move { - // First, try a simple HTTP test - debug!("dash_sdk_create_trusted: testing basic HTTP connectivity"); - match reqwest::get("https://www.google.com").await { - Ok(_) => debug!("dash_sdk_create_trusted: basic HTTP test successful (Google)"), - Err(e) => warn!(error = %e, "dash_sdk_create_trusted: basic HTTP test failed"), - } - - // Try the quorums endpoint directly - debug!("dash_sdk_create_trusted: testing quorums endpoint directly"); - match reqwest::get("https://quorums.testnet.networks.dash.org/quorums").await { - Ok(resp) => debug!(status = %resp.status(), "dash_sdk_create_trusted: direct quorums endpoint test successful"), - Err(e) => warn!(error = %e, "dash_sdk_create_trusted: direct quorums endpoint test failed"), - } - - // Now try through the provider match provider_for_prefetch.update_quorum_caches().await { Ok(_) => info!("dash_sdk_create_trusted: successfully prefetched quorums"), Err(e) => warn!(error = %e, "dash_sdk_create_trusted: failed to prefetch quorums; continuing"), diff --git a/packages/rs-sdk-trusted-context-provider/Cargo.toml b/packages/rs-sdk-trusted-context-provider/Cargo.toml index bc4110fcf5e..6107c8750ec 100644 --- a/packages/rs-sdk-trusted-context-provider/Cargo.toml +++ b/packages/rs-sdk-trusted-context-provider/Cargo.toml @@ -21,8 +21,8 @@ tracing = "0.1.41" lru = "0.16.3" arc-swap = "1.7.1" hex = "0.4.3" -futures = "0.3" url = "2.5" +tokio = { version = "1.40", features = ["rt"] } [features] default = [] @@ -47,4 +47,4 @@ token-history-contract = ["dpp/token-history-contract"] keywords-contract = ["dpp/keywords-contract"] [dev-dependencies] -tokio = { version = "1.40", features = ["macros", "rt-multi-thread"] } +tokio = { version = "1.40", features = ["macros", "rt-multi-thread", "net", "io-util"] } diff --git a/packages/rs-sdk-trusted-context-provider/src/provider.rs b/packages/rs-sdk-trusted-context-provider/src/provider.rs index 8cf73c752d7..1e47403c279 100644 --- a/packages/rs-sdk-trusted-context-provider/src/provider.rs +++ b/packages/rs-sdk-trusted-context-provider/src/provider.rs @@ -618,21 +618,54 @@ impl ContextProvider for TrustedHttpContextProvider { ))); } - // For non-WASM targets, we can use block_on to fetch + // For non-WASM targets, run the async fetch on a dedicated OS thread with its + // own tokio runtime. This avoids two classes of failure: + // + // 1. `futures::executor::block_on` deadlocks when called from inside a tokio + // runtime because reqwest's I/O needs the tokio reactor, which is blocked + // waiting for `block_on` to return. + // + // 2. `tokio::task::block_in_place` panics on current-thread runtimes. + // + // Spawning a scoped OS thread sidesteps both: the new thread has no tokio + // context, its own current-thread runtime, and borrows `self` safely via + // `std::thread::scope` (the scope is exited only after the thread joins). + // + // The overhead is acceptable because quorum cache misses are rare events. #[cfg(not(target_arch = "wasm32"))] { - // Use blocking to run async code in sync context - let quorum = - match futures::executor::block_on(self.find_quorum(quorum_type, quorum_hash)) { - Ok(q) => q, - Err(e) => { - debug!("Error finding quorum: {}", e); - return Err(ContextProviderError::Generic(format!( - "Failed to find quorum: {}", - e - ))); - } - }; + let find_future = self.find_quorum(quorum_type, quorum_hash); + let thread_result = std::thread::scope(|s| { + s.spawn(|| { + tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .map_err(|e| { + TrustedContextProviderError::NetworkError(format!( + "Failed to create tokio runtime for quorum fetch: {}", + e + )) + })? + .block_on(find_future) + }) + .join() + }); + + let quorum = match thread_result { + Ok(Ok(q)) => q, + Ok(Err(e)) => { + debug!("Error finding quorum: {}", e); + return Err(ContextProviderError::Generic(format!( + "Failed to find quorum: {}", + e + ))); + } + Err(_) => { + return Err(ContextProviderError::Generic( + "Quorum fetch thread panicked".to_string(), + )); + } + }; // Parse the public key from the 'key' field let pubkey_hex = quorum.key.trim_start_matches("0x"); @@ -816,6 +849,146 @@ impl ContextProvider for TrustedHttpContextProvider { mod tests { use super::*; + /// Regression test for https://github.com/dashpay/platform/issues/3432. + /// + /// Verifies that `get_quorum_public_key` does **not** deadlock when called + /// from inside a tokio runtime on a cache miss. + /// + /// ## Why current-thread runtime? + /// + /// The deadlock only manifests on a **current-thread** (single-threaded) + /// runtime. On a multi-thread runtime, reqwest can complete I/O on a + /// different worker thread even when the calling thread is blocked inside + /// `futures::executor::block_on`, masking the bug. + /// + /// On a current-thread runtime the single thread is both the executor and + /// the I/O driver. Calling `futures::executor::block_on` from within an + /// active tokio task parks that thread indefinitely, starving the I/O + /// driver → deadlock. + /// + /// ## Test structure + /// + /// The mock HTTP server is spun up on its own OS thread (outside the + /// current-thread runtime) so it is unaffected by what the runtime thread + /// does. `get_quorum_public_key` is called directly from an async context + /// running on the current-thread runtime. A channel with a 5-second + /// timeout detects whether the call completes or deadlocks. + #[test] + fn test_get_quorum_public_key_no_deadlock_inside_tokio_runtime() { + use std::time::Duration; + + let pubkey_bytes = [0xABu8; 48]; + let pubkey_hex = hex::encode(pubkey_bytes); + let quorum_hash: [u8; 32] = [0x01u8; 32]; + let quorum_hash_hex = hex::encode(quorum_hash); + + let current_json = serde_json::json!({ + "success": true, + "data": [{ + "quorum_hash": quorum_hash_hex, + "key": pubkey_hex, + "height": 1000u64, + "valid_members_count": 50u32, + }] + }) + .to_string(); + + let previous_json = serde_json::json!({ + "success": true, + "data": { "height": 999u64, "quorums": [] } + }) + .to_string(); + + // Spin up the mock HTTP server on a dedicated OS thread (independent of + // the current-thread runtime below so it keeps responding while the + // runtime thread is blocked). + let server_rt = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .unwrap(); + + let listener = server_rt + .block_on(tokio::net::TcpListener::bind("127.0.0.1:0")) + .unwrap(); + let server_port = listener.local_addr().unwrap().port(); + + // Shutdown signal so the server thread exits after the test completes. + let (shutdown_tx, mut shutdown_rx) = tokio::sync::oneshot::channel::<()>(); + + let server_handle = std::thread::spawn(move || { + server_rt.block_on(async move { + // Handle connections sequentially — sufficient for a test mock. + loop { + tokio::select! { + accepted = listener.accept() => { + let Ok((mut stream, _)) = accepted else { break }; + use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; + let mut reader = BufReader::new(&mut stream); + let mut request_line = String::new(); + let _ = reader.read_line(&mut request_line).await; + let body = if request_line.contains("/quorums") + && !request_line.contains("/previous") + { + current_json.clone() + } else { + previous_json.clone() + }; + let response = format!( + "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}", + body.len(), + body + ); + let _ = stream.write_all(response.as_bytes()).await; + } + _ = &mut shutdown_rx => break, + } + } + }); + }); + + let base_url = format!("http://127.0.0.1:{}", server_port); + let provider = TrustedHttpContextProvider::new_with_url( + Network::Testnet, + base_url, + NonZeroUsize::new(100).unwrap(), + ) + .expect("provider creation should succeed"); + + // Channel used to receive the result with a timeout so a deadlock is + // caught as a test failure rather than a hang. + let (tx, rx) = std::sync::mpsc::channel::>(); + + // Run get_quorum_public_key inside a current-thread tokio runtime. + // This is the exact scenario that deadlocked before the fix. + let client_handle = std::thread::spawn(move || { + let rt = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .unwrap(); + // We call the synchronous function directly from async context — + // this is what proof-verification code does in the FFI layer. + let result = rt.block_on(async move { + // Calling get_quorum_public_key here (sync fn, inside async). + // Before fix: futures::executor::block_on → deadlock. + // After fix: std::thread::scope + new runtime → no deadlock. + provider.get_quorum_public_key(1, quorum_hash, 0) + }); + let _ = tx.send(result); + }); + + let result = rx + .recv_timeout(Duration::from_secs(5)) + .expect("timed out after 5s — get_quorum_public_key likely deadlocked"); + + let key = result.expect("get_quorum_public_key should return Ok"); + assert_eq!(key, pubkey_bytes); + + // Clean up: signal server shutdown and join both threads. + let _ = shutdown_tx.send(()); + server_handle.join().expect("server thread panicked"); + client_handle.join().expect("client thread panicked"); + } + #[test] fn test_get_quorum_base_url() { assert_eq!( diff --git a/packages/rs-sdk/src/sync.rs b/packages/rs-sdk/src/sync.rs index 4cc395df613..9b84a61bc29 100644 --- a/packages/rs-sdk/src/sync.rs +++ b/packages/rs-sdk/src/sync.rs @@ -54,31 +54,72 @@ impl From for crate::Error { /// Blocks on the provided future and returns the result. /// /// This function is used to call async functions from sync code. -/// Requires the current thread to be running in a tokio runtime. /// /// Due to limitations of tokio runtime, we cannot use `tokio::runtime::Runtime::block_on` if we are already inside a tokio runtime. /// This function is a workaround for that limitation. +/// +/// Handles three scenarios: +/// - No active runtime: creates a temporary current-thread runtime and drives the future directly. +/// - Current-thread runtime: spawns a dedicated OS thread with its own independent runtime, +/// since `block_in_place` panics when there are no other worker threads. +/// - Any other runtime flavor (multi-thread, etc.): uses `block_in_place` + spawn for efficient bridging. #[cfg(not(target_arch = "wasm32"))] pub fn block_on(fut: F) -> Result where F: Future + Send + 'static, F::Output: Send, { + use tokio::runtime::RuntimeFlavor; + tracing::trace!("block_on: running async function from sync code"); - let rt = tokio::runtime::Handle::try_current()?; - let (tx, rx) = std::sync::mpsc::channel(); - tracing::trace!("block_on: Spawning worker"); - let hdl = rt.spawn(worker(fut, tx)); - tracing::trace!("block_on: Worker spawned"); - let resp = tokio::task::block_in_place(|| rx.recv())?; - - tracing::trace!("Response received"); - if !hdl.is_finished() { - tracing::debug!("async-sync worker future is not finished, aborting; this should not happen, but it's fine"); - hdl.abort(); // cleanup the worker future - } - Ok(resp) + let handle = match tokio::runtime::Handle::try_current() { + Ok(h) => h, + Err(e) => { + tracing::trace!("block_on: no active runtime ({e}), creating temporary runtime"); + return Ok(tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .map_err(|e| AsyncError::Generic(e.to_string()))? + .block_on(fut)); + } + }; + + match handle.runtime_flavor() { + RuntimeFlavor::CurrentThread => { + tracing::trace!("block_on: current-thread runtime, spawning dedicated OS thread"); + let (tx, rx) = std::sync::mpsc::sync_channel::>(1); + let join_handle = std::thread::spawn(move || { + let result = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .map_err(|e| { + tracing::error!("block_on: failed to create worker runtime: {}", e); + AsyncError::Generic(format!("failed to create worker runtime: {e}")) + }) + .map(|rt| rt.block_on(fut)); + let _ = tx.send(result); + }); + let recv_result = rx.recv()?; + join_handle + .join() + .map_err(|_| AsyncError::Generic("block_on worker thread panicked".to_string()))?; + recv_result + } + // RuntimeFlavor is #[non_exhaustive]; all multi-threaded flavors (MultiThread, + // MultiThreadAlt, and any future variants) support block_in_place. + _ => { + tracing::trace!("block_on: multi-thread runtime, using block_in_place"); + let (tx, rx) = std::sync::mpsc::channel(); + let hdl = handle.spawn(worker(fut, tx)); + let resp = tokio::task::block_in_place(|| rx.recv())?; + if !hdl.is_finished() { + tracing::debug!("async-sync worker future is not finished, aborting"); + hdl.abort(); + } + Ok(resp) + } + } } #[cfg(target_arch = "wasm32")] @@ -346,6 +387,64 @@ mod test { } } + /// Regression test for https://github.com/dashpay/platform/issues/3432. + /// + /// `block_on` previously called `tokio::task::block_in_place` unconditionally, which + /// panics on a current-thread (single-threaded) tokio runtime. The fix detects the + /// runtime flavor and spawns a dedicated OS thread with its own runtime when running + /// on a current-thread scheduler. + /// + /// This test proves the fix works: the same async-sync-async nesting that used to + /// panic now completes successfully on a current-thread runtime. + #[test] + fn test_block_on_fails_on_current_thread_runtime() { + let rt = Builder::new_current_thread() + .enable_all() + .build() + .expect("Failed to create current-thread Tokio runtime"); + + const MSGS: usize = 3; + let (tx, rx) = mpsc::channel::(1); + + let worker = async move { + for count in 0..MSGS { + tx.send(count).await.unwrap(); + } + }; + let worker_join = rt.spawn(worker); + + async fn innermost(mut rx: Receiver) -> Result { + for i in 0..MSGS { + let count = rx.recv().await.unwrap(); + assert_eq!(count, i); + } + Ok("Success".to_string()) + } + + fn sync_bridge(fut: F) -> Result + where + F: Future> + Send + 'static, + F::Output: Send, + { + block_on(fut)?.map_err(|e| ContextProviderError::Generic(e.to_string())) + } + + async fn outer(rx: Receiver) -> Result { + let result = innermost(rx).await; + sync_bridge(async { result }) + } + + let result = rt.block_on(outer(rx)); + + rt.block_on(worker_join).ok(); + + assert_eq!( + result.unwrap(), + "Success", + "block_on should succeed on a current-thread runtime" + ); + } + use crate::error::StaleNodeError; use rs_dapi_client::DapiClientError;