refactor: replace FFI OS threads with tokio tasks

[xdustinface]

Make monitoring and sync run as tokio tasks instead of dedicated OS threads. This removes the std::thread::spawn + block_on pattern and uses the existing tokio runtime directly.

Based on:

• refactor: make DashSpvClient cloneable #453

Summary by CodeRabbit

• Refactor

  • Background monitoring now runs as lightweight runtime tasks; startup returns immediately after spawning monitors and shutdown cancels/waits for them for cleaner teardown.
  • Internal handles now use task semantics for spawn/abort/wait behavior.

• Documentation

  • API docs and safety notes updated to consistently refer to "tasks" instead of "threads".

• Refactor

  • Several public callback types are now clonable for easier reuse.

[coderabbitai]

📝 Walkthrough

Walkthrough

Replaces thread-based monitoring with Tokio task-based monitoring in the SPV client, adds clonability to FFI callback structs, and updates documentation/comments to use "task" terminology. No public function signatures were changed; behavior now uses Tokio tasks and task-abort cleanup.

Changes

Cohort / File(s)	Summary
Documentation & Public Header dash-spv-ffi/FFI_API.md, dash-spv-ffi/include/dash_spv_ffi.h	Editorial updates: replaced references to "monitoring thread"/"background thread(s)" with "monitoring task"/"background task(s)" in descriptions and safety notes. No API signature changes.
Core Client Implementation dash-spv-ffi/src/client.rs	Switched runtime concurrency from std::thread to tokio::spawn tasks: active_threads → active_tasks storing JoinHandle<()>; join_active_threads() renamed to cancel_active_tasks() and now aborts/waits tasks. Run/destroy flows updated to spawn and cancel tasks; docstrings/logs updated to use "task".
FFI Callback Types dash-spv-ffi/src/callbacks.rs	Added #[derive(Clone)] to FFIProgressCallback, FFISyncEventCallbacks, FFINetworkEventCallbacks, and FFIWalletEventCallbacks enabling easy cloning of callback structs. No field or signature changes.

Sequence Diagram(s)

mermaid
sequenceDiagram
participant Client
participant Runtime as TokioRuntime
participant Monitor as MonitorTask
participant Callbacks as CallbackHandler

Client->>Runtime: spawn monitoring task(s)
Runtime-->>Monitor: run async monitor
Monitor->>Callbacks: emit events / invoke callbacks
Client->>Runtime: request cancel/abort tasks
Runtime-->>Monitor: abort and await shutdown

Estimated code review effort

🎯 3 (Moderate) | ⏱️ ~25 minutes

Poem

🐰 From threads to tasks I bound with glee,
Async fields now whisper to me.
I spawn, I call, then gently stop—
A hop, a clap, the monitors drop. 🥕

🚥 Pre-merge checks | ✅ 3

✅ Passed checks (3 passed)

Check name	Status	Explanation
Description Check	✅ Passed	Check skipped - CodeRabbit’s high-level summary is enabled.
Title check	✅ Passed	The title clearly summarizes the main change: replacing OS threads with tokio tasks across the FFI codebase.
Docstring Coverage	✅ Passed	Docstring coverage is 100.00% which is sufficient. The required threshold is 80.00%.

_{✏️ Tip: You can configure your own custom pre-merge checks in the settings.}

✨ Finishing Touches

• 📝 Generate docstrings

🧪 Generate unit tests (beta)

• Create PR with unit tests
• Post copyable unit tests in a comment
• Commit unit tests in branch refactor/replace-os-threads

---

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands and usage tips.}

[github-actions]

This PR has merge conflicts with the base branch. Please rebase or merge the base branch into your branch to resolve them.

[coderabbitai]

Actionable comments posted: 2

🧹 Nitpick comments (1)

dash-spv-ffi/src/client.rs (1)

41-41: unwrap() on mutex locks in spawned async tasks violates the library coding guideline.

Lines 41, 79, and 91 call .lock().unwrap() inside tokio::spawn bodies. A poisoned mutex would cause the task to panic; because cancel_active_tasks silently discards JoinError with let _ = task.await, the panic is swallowed and the monitoring loop exits invisibly.

The idiomatic recovery is .unwrap_or_else(|e| e.into_inner()). The same pattern applies to all other Mutex::lock().unwrap() calls in this file (lines 197, 345, 347, 358, 369, 379, 669, 684, …).

♻️ Suggested change for async task sites

-let guard = callbacks.lock().unwrap();
+let guard = callbacks.lock().unwrap_or_else(|e| e.into_inner());

As per coding guidelines: "Avoid unwrap() and expect() in library code; use explicit error types."

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@dash-spv-ffi/src/client.rs` at line 41, Replace all occurrences of
.lock().unwrap() inside spawned async tasks and other library code with a safe
recovery from PoisonError, e.g. .lock().unwrap_or_else(|e| e.into_inner()), so a
poisoned Mutex does not cause a panic that gets swallowed; specifically update
the uses such as the callbacks.lock().unwrap() inside the tokio::spawn closures
(and other Mutex::lock() calls referenced in cancel_active_tasks and the
monitoring loop) to use unwrap_or_else(|e| e.into_inner()) to recover the inner
guard instead of panicking.

🤖 Prompt for all review comments with AI agents

Verify each finding against the current code and only fix it if needed.

Inline comments:
In `@dash-spv-ffi/src/client.rs`:
- Around line 344-402: dash_spv_ffi_client_run currently can be called multiple
times and will resubscribe/spawn duplicate monitor tasks; add a running guard
(e.g., an Arc<AtomicBool> field on FFIDashSpvClient like `running`) and use an
atomic compare-and-set at the start of dash_spv_ffi_client_run to return early
if already running, set it to true when starting, and clear it in
stop_client_internal when stopping; ensure the run path and stop_client_internal
both reference this same `running` flag and that you only subscribe/spawn tasks
when the guard transitions from false->true to prevent duplicate subscriptions
and appends to `active_tasks` (keep existing calls to inner.start()/monitor
spawn logic but gate them behind the running check).
- Around line 350-388: The code is calling .resubscribe() on broadcast receivers
which discards buffered events; instead move the original receivers into
spawn_broadcast_monitor so no startup events are lost. Replace uses of
sync_event_rx.resubscribe(), network_event_rx.resubscribe(), and
wallet_event_rx.resubscribe() when spawning monitors with the original
sync_event_rx, network_event_rx, and wallet_event_rx (i.e., move them into the
spawn_broadcast_monitor calls), leaving the conditional checks and other
parameters (spawn_broadcast_monitor, spawn_progress_monitor, client.*_callbacks,
shutdown_token, client.runtime, and the closure |cb, event| cb.dispatch(event))
unchanged.

---

Nitpick comments:
In `@dash-spv-ffi/src/client.rs`:
- Line 41: Replace all occurrences of .lock().unwrap() inside spawned async
tasks and other library code with a safe recovery from PoisonError, e.g.
.lock().unwrap_or_else(|e| e.into_inner()), so a poisoned Mutex does not cause a
panic that gets swallowed; specifically update the uses such as the
callbacks.lock().unwrap() inside the tokio::spawn closures (and other
Mutex::lock() calls referenced in cancel_active_tasks and the monitoring loop)
to use unwrap_or_else(|e| e.into_inner()) to recover the inner guard instead of
panicking.

[github-actions]

This PR has merge conflicts with the base branch. Please rebase or merge the base branch into your branch to resolve them.

[coderabbitai]

Actionable comments posted: 1

🧹 Nitpick comments (1)

dash-spv-ffi/src/client.rs (1)

45-47: Silent event drop on Lagged — consider logging the skipped count.

When a broadcast receiver falls behind, Lagged(n) tells you exactly how many messages were missed. Discarding it silently makes buffer-tuning problems invisible in production.

📝 Proposed change

-                        Err(broadcast::error::RecvError::Lagged(_)) => continue,
+                        Err(broadcast::error::RecvError::Lagged(n)) => {
+                            tracing::warn!("{} monitoring task lagged, dropped {} events", name, n);
+                            continue;
+                        }

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@dash-spv-ffi/src/client.rs` around lines 45 - 47, The loop currently swallows
broadcast::error::RecvError::Lagged(_) silently; change the match arm handling
RecvError::Lagged(n) to log the dropped/skipped count instead of just continuing
so lost messages are visible in logs — locate the loop handling
Err(broadcast::error::RecvError::Lagged(_)) and replace it with a call to your
logger (e.g., processLogger, log::warn!, or the crate’s logger) that includes
the Lagged(n) value and context (which receiver/topic) and then continue; keep
the RecvError::Closed break behavior unchanged.

🤖 Prompt for all review comments with AI agents

Verify each finding against the current code and only fix it if needed.

Inline comments:
In `@dash-spv-ffi/src/client.rs`:
- Around line 40-43: The code currently calls dispatch_fn while holding the
Mutex guarding callbacks (the callbacks variable), which can deadlock or starve
tokio workers if the C callback blocks or re-enters FFI setters like
dash_spv_ffi_client_set_sync_event_callbacks /
dash_spv_ffi_client_clear_sync_event_callbacks or invokes stop/destroy paths
that call cancel_active_tasks → runtime.block_on; fix by cloning the
Option-contained callback out of the mutex before releasing the lock (add C:
Clone where needed) and then call dispatch_fn after the lock is dropped, or
alternatively offload callback invocation to a blocking thread via
tokio::task::spawn_blocking to avoid holding the async worker; apply the same
change to the progress-monitor callback paths (the other callback mutex usage)
and update the Safety docs for callback setters to forbid acquiring the same
mutex from within callbacks.

---

Duplicate comments:
In `@dash-spv-ffi/src/client.rs`:
- Around line 311-401: The run function can be called twice and will append
duplicate monitoring tasks to active_tasks; add a guard at the top of
dash_spv_ffi_client_run that checks client.active_tasks (lock and inspect) and
if it is non-empty return an appropriate error (use set_last_error and return an
FFI error code like the earlier FFIErrorCode conversion) instead of spawning new
tasks, or alternatively add/inspect a dedicated "started" boolean on the client
(e.g., client.started) that you set after successful start and check before
subscribing/spawning to prevent duplicate subscriptions and task creation;
ensure you hold the same mutex (active_tasks or the new flag) consistently when
checking/setting to avoid races.

---

Nitpick comments:
In `@dash-spv-ffi/src/client.rs`:
- Around line 45-47: The loop currently swallows
broadcast::error::RecvError::Lagged(_) silently; change the match arm handling
RecvError::Lagged(n) to log the dropped/skipped count instead of just continuing
so lost messages are visible in logs — locate the loop handling
Err(broadcast::error::RecvError::Lagged(_)) and replace it with a call to your
logger (e.g., processLogger, log::warn!, or the crate’s logger) that includes
the Lagged(n) value and context (which receiver/topic) and then continue; keep
the RecvError::Closed break behavior unchanged.

[coderabbitai]

🧹 Nitpick comments (2)

dash-spv-ffi/src/client.rs (2)

45-46: Add a warning log when broadcast events are dropped due to receiver lag.

Silent continue on RecvError::Lagged makes it invisible in production when the callback dispatch path can't keep up with the event rate — a condition that directly affects callback completeness.

💡 Suggested fix

-                        Err(broadcast::error::RecvError::Lagged(_)) => continue,
+                        Err(broadcast::error::RecvError::Lagged(n)) => {
+                            tracing::warn!("{} monitoring task lagged, {} events dropped", name, n);
+                            continue;
+                        }

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@dash-spv-ffi/src/client.rs` around lines 45 - 46, The branch that silently
continues on Err(broadcast::error::RecvError::Lagged(_)) should log a warning so
dropped events are visible; change the pattern to capture the dropped count
(e.g. Err(broadcast::error::RecvError::Lagged(dropped))) and emit a warning
(using your project's logger or tracing::warn!/log::warn!) that includes the
dropped value and context (e.g. which callback or task) before continuing, then
keep the continue. This targets the recv error handling branch where
RecvError::Lagged is matched.

---

45-46: Consider logging a warning when broadcast events are dropped due to receiver lag.

RecvError::Lagged means the consumer fell behind and some events were silently discarded. Without a log entry, this is invisible in production and makes callback completeness hard to diagnose.

💡 Suggested fix

-                        Err(broadcast::error::RecvError::Lagged(_)) => continue,
+                        Err(broadcast::error::RecvError::Lagged(n)) => {
+                            tracing::warn!("{} monitoring task lagged, dropped {} events", name, n);
+                            continue;
+                        }

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@dash-spv-ffi/src/client.rs` around lines 45 - 46, The match arm handling
broadcast::error::RecvError::Lagged currently silently continues; change it to
log a warning including the dropped count before continuing so missed events are
visible in production. In the match on broadcast::error::RecvError (the arm with
RecvError::Lagged(_)), bind the count (e.g., RecvError::Lagged(missed)) and emit
a warning via your logging/tracing facility (include missed) and then continue;
leave the RecvError::Closed behavior unchanged.

🤖 Prompt for all review comments with AI agents

Verify each finding against the current code and only fix it if needed.

Duplicate comments:
In `@dash-spv-ffi/src/client.rs`:
- Line 349: The current use of .resubscribe() on sync_event_rx,
network_event_rx, and wallet_event_rx drops any events received between the
initial subscribe calls and the resubscribe, so change spawn_broadcast_monitor
invocations to accept and use the original receivers (sync_event_rx,
network_event_rx, wallet_event_rx) obtained inside the block_on rather than
calling .resubscribe() there; locate the spawn_broadcast_monitor calls and pass
the existing receiver variables directly so the spawned tasks receive the
existing buffered events (e.g., SyncStart) instead of creating new
tail-positioned receivers.
- Around line 311-405: dash_spv_ffi_client_run currently starts the client and
pushes monitoring tasks every call without guarding against repeated runs; add a
guard that returns an error if the client is already running (or otherwise
no-op) by checking a running flag or whether client.active_tasks (or a new
client.is_running AtomicBool) indicates tasks already exist before starting; set
this flag to true immediately after successful start and to false in
dash_spv_ffi_client_stop (or when tasks finish/are cleared), and ensure spawn of
monitors (spawn_broadcast_monitor/spawn_progress_monitor and the runtime.spawn
monitor_network task) only occurs when not already running to prevent duplicate
subscriptions and dispatches.
- Around line 194-210: cancel_active_tasks currently calls self.runtime.block_on
which will panic if invoked from within a tokio worker task (e.g., when
callbacks dispatched from tokio tasks call dash_spv_ffi_client_stop or
dash_spv_ffi_client_destroy); update the code and docs: add explicit safety doc
comments to all callback setter APIs stating callbacks MUST NOT call
dash_spv_ffi_client_stop or dash_spv_ffi_client_destroy because
Runtime::block_on will panic when called inside a tokio runtime, and for a
structural fix change how callbacks are dispatched so they run on a blocking OS
thread (use tokio::task::spawn_blocking or otherwise offload callbacks to a
dedicated blocking thread) so that cancel_active_tasks can safely call
Runtime::block_on; reference cancel_active_tasks, dash_spv_ffi_client_stop,
dash_spv_ffi_client_destroy, and the callback setter functions when applying
these changes.
- Around line 311-405: The run function (dash_spv_ffi_client_run) lacks a guard
against sequential calls so repeated calls start inner.start(), re-subscribe and
push duplicate monitors into client.active_tasks; fix by checking a run-state
before starting (e.g. inspect client.active_tasks.lock().unwrap().is_empty() or
add an AtomicBool like client.is_running) and if already running return a
suitable error (FFIErrorCode::AlreadyRunning as i32) instead of proceeding; only
call client.inner.start(), subscribe to channels, and push monitors when the
guard shows the client is not already running, and set the run-state (or mark
active_tasks) after spawning tasks so subsequent calls are no-ops or fail fast.
- Line 349: The current code calls .resubscribe() on sync_event_rx,
network_event_rx, and wallet_event_rx which creates new receivers at the current
tail and drops any events buffered by the original subscribe_*() calls; instead,
move the original receivers into spawn_broadcast_monitor so those buffered
startup events are preserved. Concretely: stop calling .resubscribe() where
sync_event_rx.resubscribe(), network_event_rx.resubscribe(),
wallet_event_rx.resubscribe() are used; pass the existing sync_event_rx,
network_event_rx, wallet_event_rx variables into spawn_broadcast_monitor (and
update spawn_broadcast_monitor's parameter types to accept the broadcast
Receiver types returned by subscribe_*()); remove the resubscribe calls and any
cloning that discards buffered events.
- Around line 194-210: The cancel_active_tasks function currently calls
self.runtime.block_on(...) which panics when invoked on a tokio worker thread
(as happens from a C callback path like dash_spv_ffi_client_stop →
stop_client_internal → cancel_active_tasks); change the implementation to run
the blocking abort-and-wait logic off the async worker by using
tokio::task::spawn_blocking (or otherwise dispatch to a dedicated std::thread)
to perform task.abort() for each active task and await their completion outside
the runtime, and update any callback setter docs (and callers like stop() /
destroy()) to forbid invoking stop/destroy from inside a callback if you keep
the blocking approach; refer to cancel_active_tasks, self.runtime.block_on,
task.abort, spawn_blocking, dash_spv_ffi_client_stop, stop_client_internal,
stop(), and destroy() when making the change.

---

Nitpick comments:
In `@dash-spv-ffi/src/client.rs`:
- Around line 45-46: The branch that silently continues on
Err(broadcast::error::RecvError::Lagged(_)) should log a warning so dropped
events are visible; change the pattern to capture the dropped count (e.g.
Err(broadcast::error::RecvError::Lagged(dropped))) and emit a warning (using
your project's logger or tracing::warn!/log::warn!) that includes the dropped
value and context (e.g. which callback or task) before continuing, then keep the
continue. This targets the recv error handling branch where RecvError::Lagged is
matched.
- Around line 45-46: The match arm handling broadcast::error::RecvError::Lagged
currently silently continues; change it to log a warning including the dropped
count before continuing so missed events are visible in production. In the match
on broadcast::error::RecvError (the arm with RecvError::Lagged(_)), bind the
count (e.g., RecvError::Lagged(missed)) and emit a warning via your
logging/tracing facility (include missed) and then continue; leave the
RecvError::Closed behavior unchanged.

[ZocoLini]

I think that the next step should be moving this logic into dash-spv client and make the ffi wrap the methods defined there, that way we avoid Rust user from having to monitor a reciever, making it 'automatic' for all user and not only for FFI users