fix(platform-wallet): enforce FFI lifetime invariants

packages/rs-platform-wallet/src/manager/mod.rs

@@ -64,6 +64,9 @@ pub struct PlatformWalletManager<P: PlatformWalletPersistence + 'static> {
     /// auto-started — call `start` after wallets are registered. See
     /// [`DashPaySyncManager`].
     pub(super) dashpay_sync_manager: Arc<DashPaySyncManager>,
+    /// Tracks asynchronous payment hooks so manager shutdown can close
+    /// admission and drain every task before host callback contexts are freed.
+    pub(super) dashpay_payment_handler: Arc<DashPayPaymentHandler>,
     /// Periodic shielded (Orchard) note sync coordinator (spends are
     /// detected during the note scan, no separate nullifier pass).
     /// Iterates every wallet that has been bound via
@@ -170,7 +173,7 @@ impl<P: PlatformWalletPersistence + 'static> PlatformWalletManager<P> {
             app_handler,
             lock_handler,
             balance_handler,
-            dashpay_payment_handler,
+            Arc::clone(&dashpay_payment_handler) as Arc<dyn PlatformEventHandler>,
         ]));
 
         let spv = Arc::new(SpvRuntime::new(
@@ -206,6 +209,7 @@ impl<P: PlatformWalletPersistence + 'static> PlatformWalletManager<P> {
             platform_address_sync_manager: platform_address_sync,
             identity_sync_manager: identity_sync,
             dashpay_sync_manager: dashpay_sync,
+            dashpay_payment_handler,
             #[cfg(feature = "shielded")]
             shielded_sync_manager: shielded_sync,
             #[cfg(feature = "shielded")]
@@ -427,7 +431,7 @@ impl<P: PlatformWalletPersistence + 'static> PlatformWalletManager<P> {
 
     /// Stop all background tasks and wait for them to exit.
     ///
-    /// **Quiesces** the periodic coordinators
+    /// Stops SPV and **quiesces** the periodic coordinators
     /// (`PlatformAddressSyncManager`, `IdentitySyncManager`,
     /// `DashPaySyncManager`, `ShieldedSyncManager`) — cancelling each
     /// loop *and draining any in-flight pass to completion*, including
@@ -437,7 +441,9 @@ impl<P: PlatformWalletPersistence + 'static> PlatformWalletManager<P> {
     /// shutdown is required (e.g. on app termination); a dirty drop
     /// simply leaks the tasks until the runtime exits.
     ///
-    /// Ordering matters: cancel-only `stop()` would let a pass already
+    /// Ordering matters: SPV is stopped and joined first so it cannot dispatch
+    /// more wallet events. Payment-task admission is then closed and all
+    /// admitted work is joined. A cancel-only `stop()` would let a pass already
     /// inside `sync_now` keep running and call `persister.store(...)` /
     /// fire a host completion callback after the FFI's `destroy`
     /// returned and the host freed the persister / event-handler
@@ -446,6 +452,11 @@ impl<P: PlatformWalletPersistence + 'static> PlatformWalletManager<P> {
     /// and only THEN cancel + join the event adapter, which is the sink
     /// those stores feed into.
     pub async fn shutdown(&self) {
+        if let Err(error) = self.spv_manager.stop().await {
+            tracing::warn!(?error, "SPV shutdown failed");
+        }
+
+        self.dashpay_payment_handler.quiesce().await;
         self.platform_address_sync_manager.quiesce().await;
         self.identity_sync_manager.quiesce().await;
         self.dashpay_sync_manager.quiesce().await;

packages/rs-platform-wallet/src/spv/runtime.rs

@@ -1,6 +1,7 @@
 //! SPV client runtime — manages the DashSpvClient lifecycle.
 
 use std::sync::{Arc, Mutex};
+use std::time::Duration;
 
 use tokio::sync::RwLock;
 use tokio::task::JoinHandle;
@@ -24,6 +25,28 @@ use crate::wallet::platform_wallet::PlatformWalletInfo;
 type SpvClient =
     DashSpvClient<WalletManager<PlatformWalletInfo>, PeerNetworkManager, DiskStorageManager>;
 
+const SPV_STOP_TIMEOUT: Duration = Duration::from_secs(15);
+
+/// Join a stopped SPV runner, escalating to cancellation after `timeout` but
+/// never returning until Tokio confirms that the task has terminated.
+async fn join_spv_task(mut handle: JoinHandle<()>, timeout: Duration) {
+    match tokio::time::timeout(timeout, &mut handle).await {
+        Ok(Ok(())) => {}
+        Ok(Err(error)) => {
+            tracing::warn!(?error, "SPV background run loop join error");
+        }
+        Err(_) => {
+            tracing::warn!("SPV stop: background run loop did not unwind in time; aborting it");
+            handle.abort();
+            if let Err(error) = handle.await {
+                if !error.is_cancelled() {
+                    tracing::warn!(?error, "SPV background run loop abort join error");
+                }
+            }
+        }
+    }
+}
+
 /// SPV client runtime — owns the `DashSpvClient` and drives sync.
 ///
 /// Events are dispatched through [`PlatformEventManager`] to all registered
@@ -217,17 +240,7 @@ impl SpvRuntime {
 
         let handle = self.task.lock().expect("spv task mutex poisoned").take();
         if let Some(handle) = handle {
-            let abort = handle.abort_handle();
-            if tokio::time::timeout(std::time::Duration::from_secs(15), handle)
-                .await
-                .is_err()
-            {
-                tracing::warn!(
-                    "SPV stop: background run loop did not unwind within 15s; aborting it"
-                );
-
-                abort.abort();
-            }
+            join_spv_task(handle, SPV_STOP_TIMEOUT).await;
         }
 
         stop_result
@@ -415,6 +428,40 @@ impl SpvRuntime {
     }
 }
 
+#[cfg(test)]
+mod shutdown_tests {
+    use super::*;
+    use std::sync::atomic::{AtomicBool, Ordering};
+
+    struct DropFlag(Arc<AtomicBool>);
+
+    impl Drop for DropFlag {
+        fn drop(&mut self) {
+            self.0.store(true, Ordering::SeqCst);
+        }
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn timed_out_spv_task_is_aborted_and_joined_before_return() {
+        let dropped = Arc::new(AtomicBool::new(false));
+        let dropped_in_task = Arc::clone(&dropped);
+        let (started_tx, started_rx) = tokio::sync::oneshot::channel();
+        let handle = tokio::spawn(async move {
+            let _flag = DropFlag(dropped_in_task);
+            let _ = started_tx.send(());
+            std::future::pending::<()>().await;
+        });
+        started_rx.await.expect("SPV task should start");
+
+        join_spv_task(handle, SPV_STOP_TIMEOUT).await;
+
+        assert!(
+            dropped.load(Ordering::SeqCst),
+            "abort must be joined so task-owned callback state is dropped"
+        );
+    }
+}
+
 impl std::fmt::Debug for SpvRuntime {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         f.debug_struct("SpvRuntime")

packages/rs-platform-wallet/src/wallet/identity/network/payment_handler.rs

@@ -24,6 +24,7 @@
 //! to a core transaction row.
 
 use std::sync::Arc;
+use std::{future::Future, sync::Mutex};
 
 use dash_spv::EventHandler;
 use key_wallet::managed_account::transaction_record::TransactionRecord;
@@ -44,6 +45,70 @@ use crate::wallet::platform_wallet::PlatformWalletInfo;
 pub(crate) struct DashPayPaymentHandler {
     wallet_manager: Arc<RwLock<WalletManager<PlatformWalletInfo>>>,
     persister: Arc<dyn PlatformWalletPersistence>,
+    tasks: PaymentTaskTracker,
+}
+
+struct PaymentTaskState {
+    accepting: bool,
+    handles: Vec<tokio::task::JoinHandle<()>>,
+}
+
+/// Admission gate and join barrier for asynchronous payment hooks.
+///
+/// Event dispatch is synchronous, so the gate uses a short standard mutex:
+/// checking admission and recording the resulting task happen atomically with
+/// respect to shutdown. No mutex guard is held while a task is awaited.
+struct PaymentTaskTracker {
+    state: Mutex<PaymentTaskState>,
+}
+
+impl PaymentTaskTracker {
+    fn new() -> Self {
+        Self {
+            state: Mutex::new(PaymentTaskState {
+                accepting: true,
+                handles: Vec::new(),
+            }),
+        }
+    }
+
+    fn spawn<F>(&self, future: F) -> bool
+    where
+        F: Future<Output = ()> + Send + 'static,
+    {
+        let mut state = self.state.lock().expect("payment task mutex poisoned");
+        if !state.accepting {
+            return false;
+        }
+
+        // Completed tasks cannot fire another callback and need not be kept
+        // until manager destruction. This bounds the tracker during long syncs.
+        state.handles.retain(|handle| !handle.is_finished());
+        state.handles.push(tokio::spawn(future));
+        true
+    }
+
+    async fn quiesce(&self) {
+        let handles = {
+            let mut state = self.state.lock().expect("payment task mutex poisoned");
+            state.accepting = false;
+            std::mem::take(&mut state.handles)
+        };
+
+        for handle in handles {
+            if let Err(error) = handle.await {
+                tracing::warn!(?error, "DashPay payment task join error");
+            }
+        }
+    }
+
+    #[cfg(test)]
+    fn is_accepting(&self) -> bool {
+        self.state
+            .lock()
+            .expect("payment task mutex poisoned")
+            .accepting
+    }
 }
 
 impl DashPayPaymentHandler {
@@ -54,8 +119,15 @@ impl DashPayPaymentHandler {
         Self {
             wallet_manager,
             persister,
+            tasks: PaymentTaskTracker::new(),
         }
     }
+
+    /// Stop admitting callback-bearing work and join every task admitted
+    /// before the gate closed. Idempotent.
+    pub(crate) async fn quiesce(&self) {
+        self.tasks.quiesce().await;
+    }
 }
 
 impl EventHandler for DashPayPaymentHandler {
@@ -69,7 +141,7 @@ impl EventHandler for DashPayPaymentHandler {
         let wallet_manager = Arc::clone(&self.wallet_manager);
         let persister = Arc::clone(&self.persister);
         let event = event.clone();
-        tokio::spawn(async move {
+        self.tasks.spawn(async move {
             let wallet_id = event.wallet_id();
             let wallet_persister =
                 crate::wallet::persister::WalletPersister::new(wallet_id, persister);
@@ -185,6 +257,7 @@ mod tests {
     use key_wallet::managed_account::transaction_record::TransactionDirection;
     use key_wallet::transaction_checking::{TransactionContext, TransactionType};
     use key_wallet::WalletCoreBalance;
+    use std::sync::atomic::{AtomicBool, Ordering};
 
     /// A `TransactionRecord` whose txid is uniquely seeded by `seed` (via a
     /// distinct input outpoint). Context is irrelevant to the routing under
@@ -323,4 +396,47 @@ mod tests {
         assert!(dashpay_payment_records(&event).is_empty());
         assert!(!drives_payment_hooks(&event));
     }
+
+    #[tokio::test]
+    async fn payment_task_quiesce_closes_admission_and_joins_in_flight_work() {
+        let tasks = Arc::new(PaymentTaskTracker::new());
+        let (started_tx, started_rx) = tokio::sync::oneshot::channel();
+        let (release_tx, release_rx) = tokio::sync::oneshot::channel();
+        let completed = Arc::new(AtomicBool::new(false));
+        let completed_by_task = Arc::clone(&completed);
+
+        assert!(tasks.spawn(async move {
+            let _ = started_tx.send(());
+            let _ = release_rx.await;
+            completed_by_task.store(true, Ordering::SeqCst);
+        }));
+        started_rx.await.expect("tracked task should start");
+
+        let tasks_for_shutdown = Arc::clone(&tasks);
+        let shutdown = tokio::spawn(async move {
+            tasks_for_shutdown.quiesce().await;
+        });
+
+        while tasks.is_accepting() {
+            tokio::task::yield_now().await;
+        }
+        assert!(!shutdown.is_finished(), "quiesce must await admitted work");
+
+        let ran_after_close = Arc::new(AtomicBool::new(false));
+        let ran_after_close_in_task = Arc::clone(&ran_after_close);
+        assert!(!tasks.spawn(async move {
+            ran_after_close_in_task.store(true, Ordering::SeqCst);
+        }));
+
+        release_tx
+            .send(())
+            .expect("tracked task should still exist");
+        shutdown.await.expect("quiesce task should join");
+        assert!(completed.load(Ordering::SeqCst));
+        assert!(!ran_after_close.load(Ordering::SeqCst));
+
+        // Repeated shutdown is safe and keeps admission closed.
+        tasks.quiesce().await;
+        assert!(!tasks.spawn(async {}));
+    }
 }