fedimint_core/
runtime.rs

1//! Abstraction over an executor so we can spawn tasks under WASM the same way
2//! we do usually.
3
4use std::future::Future;
5
6use fedimint_logging::LOG_RUNTIME;
7pub use n0_future::task::{JoinError, JoinHandle};
8pub use n0_future::time::{Duration, Elapsed, Instant, sleep, sleep_until, timeout};
9use tracing::{Instrument, Span};
10
11use crate::task::MaybeSend;
12
13pub fn spawn<F, T>(name: &str, future: F) -> JoinHandle<T>
14where
15    F: Future<Output = T> + 'static + MaybeSend,
16    T: MaybeSend + 'static,
17{
18    let span = tracing::debug_span!(target: LOG_RUNTIME, parent: None, "spawn", task = name);
19    n0_future::task::spawn(future.instrument(span))
20}
21
22/// Like [`spawn`] but with an explicit parent span.
23///
24/// Events from the spawned future inherit fields from `parent` (e.g. `fed_id`
25/// from the client span), including the lifecycle events emitted by
26/// [`crate::task::TaskGroup`] around the user future.
27pub fn spawn_with_span<F, T>(parent: &Span, name: &str, future: F) -> JoinHandle<T>
28where
29    F: Future<Output = T> + 'static + MaybeSend,
30    T: MaybeSend + 'static,
31{
32    let span = tracing::debug_span!(target: LOG_RUNTIME, parent: parent, "spawn", task = name);
33    n0_future::task::spawn(future.instrument(span))
34}
35
36// Note: These functions only exist on non-wasm platforms and you need to handle
37// them conditionally at the call site of packages that compile on wasm
38#[cfg(not(target_family = "wasm"))]
39pub fn block_in_place<F, R>(f: F) -> R
40where
41    F: FnOnce() -> R,
42{
43    // nosemgrep: ban-raw-block-in-place
44    tokio::task::block_in_place(f)
45}
46
47#[cfg(not(target_family = "wasm"))]
48pub fn block_on<F: Future>(future: F) -> F::Output {
49    // nosemgrep: ban-raw-block-on
50    tokio::runtime::Handle::current().block_on(future)
51}
52
53/// The single, process-wide multi-threaded Tokio runtime that backs every
54/// Fedimint UniFFI entry point.
55///
56/// UniFFI's `#[uniffi::export(async_runtime = "tokio")]` async methods are
57/// polled by `async_compat::Compat` on whatever host thread (a Kotlin
58/// coroutine dispatcher, a Swift `Task`, ...) drives the call. That thread
59/// usually has no ambient Tokio runtime, so `async_compat` enters its own
60/// global *current-thread* runtime. Any work *spawned* from that context
61/// therefore runs on a current-thread runtime, and Fedimint's rocksdb access
62/// uses [`block_in_place`], which aborts the process with "can call blocking
63/// only when running on the multi-threaded runtime" when run inside such a
64/// task.
65///
66/// Routing all spawned FFI work onto this dedicated multi-threaded runtime
67/// keeps [`block_in_place`] valid regardless of which host thread drove the
68/// poll. It is created once, lazily, and reused for the life of the process,
69/// so this is a fixed worker pool — not a thread per call.
70#[cfg(all(feature = "uniffi", not(target_family = "wasm")))]
71pub fn ffi_runtime() -> &'static tokio::runtime::Runtime {
72    use std::sync::OnceLock;
73    static RUNTIME: OnceLock<tokio::runtime::Runtime> = OnceLock::new();
74    RUNTIME.get_or_init(|| {
75        tokio::runtime::Builder::new_multi_thread()
76            .enable_all()
77            .thread_name("fedimint-ffi")
78            .build()
79            .expect("failed to build fedimint FFI runtime")
80    })
81}
82
83/// Run `fut` to completion on the shared multi-threaded [`ffi_runtime`],
84/// regardless of which (possibly current-thread) runtime is ambient on the
85/// thread driving the FFI poll.
86///
87/// Wrap the body of a `#[uniffi::export(async_runtime = "tokio")]` async
88/// method in this whenever its work — or anything it transitively spawns,
89/// such as a client's state-machine executor — must run on genuine
90/// multi-threaded workers.
91#[cfg(all(feature = "uniffi", not(target_family = "wasm")))]
92pub async fn ffi_spawn<F, T>(fut: F) -> T
93where
94    F: Future<Output = T> + Send + 'static,
95    T: Send + 'static,
96{
97    ffi_runtime()
98        .spawn(fut)
99        .await
100        .expect("fedimint FFI task panicked")
101}
102
103/// Spawn a detached background task that drives a UniFFI `subscribe_*`
104/// callback loop on the shared multi-threaded [`ffi_runtime`].
105///
106/// This is only meant for the fire-and-forget background loops behind
107/// `subscribe_*` FFI methods; the spawned task is detached. See
108/// [`ffi_runtime`] for why a multi-threaded runtime is required here.
109#[cfg(all(feature = "uniffi", not(target_family = "wasm")))]
110pub fn ffi_spawn_subscription<F>(name: &str, future: F)
111where
112    F: Future<Output = ()> + Send + 'static,
113{
114    let span = tracing::debug_span!(target: LOG_RUNTIME, parent: None, "spawn", task = name);
115    ffi_runtime().spawn(future.instrument(span));
116}
117
118/// wasm has no multi-threaded runtime and no [`block_in_place`] (rocksdb is
119/// not used there), so the ambient-runtime hazard does not exist: just run
120/// the future on the ambient executor like [`spawn`] does.
121#[cfg(all(feature = "uniffi", target_family = "wasm"))]
122pub async fn ffi_spawn<F: Future>(fut: F) -> F::Output {
123    fut.await
124}
125
126#[cfg(all(feature = "uniffi", target_family = "wasm"))]
127pub fn ffi_spawn_subscription<F>(name: &str, future: F)
128where
129    F: Future<Output = ()> + 'static,
130{
131    spawn(name, future);
132}
fedimint_core/runtime.rs

fedimint_core/
runtime.rs
