fix(spv): apply InstantSend locks to self-broadcast transactions

src/backend_task/core/mod.rs

@@ -33,7 +33,7 @@ use dash_sdk::dpp::key_wallet::wallet::managed_wallet_info::transaction_builder:
     BuilderError, TransactionBuilder,
 };
 use dash_sdk::dpp::key_wallet::wallet::managed_wallet_info::wallet_info_interface::WalletInfoInterface;
-use dash_sdk::dpp::key_wallet_manager::{WalletError, WalletId, WalletManager};
+use dash_sdk::dpp::key_wallet_manager::{WalletError, WalletId, WalletInterface, WalletManager};
 use std::path::PathBuf;
 use std::str::FromStr;
 use std::sync::{Arc, RwLock};
@@ -593,7 +593,13 @@ impl AppContext {
                 &parsed_recipients,
                 &request,
             )?;
-            self.sign_spv_transaction(&mut wm, &wallet_id, unsigned)?
+            let signed = self.sign_spv_transaction(&mut wm, &wallet_id, unsigned)?;
+
+            // Notify the wallet about the outgoing tx while still holding the
+            // write lock. This marks spent UTXOs immediately so concurrent
+            // callers don't select the same inputs (double-spend prevention).
+            let _ = wm.process_mempool_transaction(&signed, false).await;
+            signed
         };
 
         self.spv_manager

src/spv/manager.rs

@@ -144,6 +144,13 @@ pub(crate) struct SpvEventHandler {
     connection_status: Option<Arc<ConnectionStatus>>,
     reconcile_tx: Arc<Mutex<Option<mpsc::Sender<()>>>>,
     finality_tx: Arc<Mutex<Option<mpsc::Sender<AssetLockFinalityEvent>>>>,
+    /// Wallet manager reference for applying InstantSend locks directly.
+    ///
+    /// Self-broadcast transactions bypass the MempoolManager and are fed to
+    /// the WalletManager via `notify_wallet_after_broadcast()`. When the IS
+    /// lock arrives later, the MempoolManager doesn't know about the tx and
+    /// cannot apply the lock. We apply it here directly on the WalletManager.
+    wallet: Arc<AsyncRwLock<WalletManager<ManagedWalletInfo>>>,
 }
 
 impl EventHandler for SpvEventHandler {
@@ -256,6 +263,33 @@ impl EventHandler for SpvEventHandler {
                 | SyncEvent::SyncComplete { .. }
         );
 
+        // TODO(workaround): Remove once dashpay/rust-dashcore#487 is fixed.
+        //
+        // Apply InstantSend locks directly on the WalletManager.
+        //
+        // Self-broadcast transactions bypass the MempoolManager (they are fed
+        // directly to WalletManager via notify_wallet_after_broadcast — see
+        // the other workaround in spawn_request_handler). When the IS lock
+        // arrives from the network, the MempoolManager doesn't know about
+        // the tx and stores it as a "pending IS lock" that is never matched.
+        // Applying the lock here ensures self-broadcast txs transition from
+        // unconfirmed to spendable.
+        //
+        // Once upstream broadcast calls handle_tx() on the MempoolManager,
+        // both workarounds (notify_wallet_after_broadcast and this) can be
+        // removed — the normal MempoolManager pipeline will handle everything.
+        //
+        // For MempoolManager-tracked txs this is a harmless no-op — the
+        // WalletManager deduplicates via its instant_send_locks HashSet.
+        if let SyncEvent::InstantLockReceived { instant_lock, .. } = event {
+            let txid = instant_lock.txid;
+            let wallet = Arc::clone(&self.wallet);
+            tokio::spawn(async move {
+                let mut wm = wallet.write().await;
+                wm.process_instant_send_lock(txid);
+            });
+        }
+
         // Forward finality-relevant events for asset lock proof construction.
         let finality_tx = self.finality_tx.lock().ok().and_then(|g| g.clone());
         if let Some(ref ftx) = finality_tx {
@@ -1379,6 +1413,7 @@ impl SpvManager {
             connection_status: self.connection_status_snapshot(),
             reconcile_tx: Arc::clone(&self.reconcile_tx),
             finality_tx: Arc::clone(&self.finality_tx),
+            wallet: Arc::clone(&self.wallet),
         });
 
         DashSpvClient::new(

tests/backend-e2e/framework/cleanup.rs

@@ -50,6 +50,9 @@ pub async fn cleanup_test_wallets(
         wallet_hashes.len()
     );
 
+    let mut swept = 0u32;
+    let mut deleted = 0u32;
+
     for hash in wallet_hashes {
         let wallet_arc = {
             let wallets = app_context.wallets().read().expect("wallets lock");
@@ -61,25 +64,43 @@ pub async fn cleanup_test_wallets(
 
         // Wait briefly for SPV to sync this wallet's balance.
         let _ =
-            wait::wait_for_spendable_balance(app_context, hash, 1, Duration::from_secs(10)).await;
+            wait::wait_for_spendable_balance(app_context, hash, 1, Duration::from_secs(1)).await;
 
-        let balance = {
+        let (spendable, total) = {
             let wallet = wallet_arc.read().expect("wallet lock");
-            wallet.confirmed_balance_duffs()
+            (
+                wallet.confirmed_balance_duffs(),
+                wallet.total_balance_duffs(),
+            )
         };
 
-        if balance == 0 {
+        // Delete wallets with no funds at all — they're fully spent orphans
+        // from previous runs. Without this, wallets accumulate across runs
+        // and degrade performance (~10 MB + 185 monitored addresses each).
+        if total == 0 {
+            if let Err(e) = app_context.remove_wallet(&hash) {
+                tracing::warn!(
+                    "Cleanup: failed to delete empty wallet {:?}: {}",
+                    &hash[..4],
+                    e
+                );
+            } else {
+                deleted += 1;
+            }
             continue;
         }
 
-        // TODO(CMT-032): Also withdraw Platform credits from test identities back to
-        // the framework wallet. Requires: enumerate identities owned by test wallets,
-        // call IdentityTask::WithdrawFromIdentity for each, wait for Core balance.
+        if spendable == 0 {
+            // Has unconfirmed funds but nothing spendable — skip sweep,
+            // will be cleaned up on a future run once funds confirm.
+            continue;
+        }
 
+        // Attempt to sweep spendable funds back to framework wallet
         let request = WalletPaymentRequest {
             recipients: vec![PaymentRecipient {
                 address: framework_address.clone(),
-                amount_duffs: balance,
+                amount_duffs: spendable,
             }],
             subtract_fee_from_amount: true,
             memo: Some("E2E cleanup: sweep orphaned wallet".to_string()),
@@ -92,12 +113,19 @@ pub async fn cleanup_test_wallets(
         });
 
         match run_task(app_context, task).await {
-            Ok(_) => tracing::info!(
-                "Cleanup: returned {} duffs from orphaned wallet {:?}",
-                balance,
-                &hash[..4]
-            ),
+            Ok(_) => {
+                swept += 1;
+                tracing::info!(
+                    "Cleanup: returned {} duffs from orphaned wallet {:?}",
+                    spendable,
+                    &hash[..4]
+                );
+            }
             Err(e) => tracing::warn!("Cleanup: failed to sweep wallet {:?}: {}", &hash[..4], e),
         }
     }
+
+    if swept > 0 || deleted > 0 {
+        tracing::info!("Cleanup complete: {swept} swept, {deleted} deleted (empty)");
+    }
 }

tests/backend-e2e/framework/harness.rs

@@ -210,13 +210,24 @@ impl BackendTestContext {
             .await
             .expect("Framework wallet not picked up by SPV");
 
-        // Wait for SPV to sync and funds to become spendable
+        // Wait for SPV to fully sync (including masternodes) so MempoolManager
+        // is active and bloom filter is built before any test broadcasts.
+        // This must come BEFORE the spendable balance check — wallet balances
+        // are only available after compact filter sync completes.
+        tracing::info!("Waiting for SPV to complete full sync (masternodes + mempool)...");
+        wait::wait_for_spv_running(&app_context, Duration::from_secs(300))
+            .await
+            .expect("SPV did not reach Running state within 300s");
+        tracing::info!("SPV fully synced — mempool bloom filter active");
+
+        // Now check framework wallet balance — SPV has synced, so balances
+        // should be available immediately (no need for a long timeout).
         tracing::info!("Waiting for SPV to sync framework wallet spendable balance...");
         match wait::wait_for_spendable_balance(
             &app_context,
             framework_wallet_hash,
             1, // at least 1 duff spendable
-            Duration::from_secs(180),
+            Duration::from_secs(30),
         )
         .await
         {
@@ -249,14 +260,6 @@ impl BackendTestContext {
             }
         }
 
-        // Wait for SPV to fully sync (including masternodes) so MempoolManager
-        // is active and bloom filter is built before any test broadcasts.
-        tracing::info!("Waiting for SPV to complete full sync (masternodes + mempool)...");
-        wait::wait_for_spv_running(&app_context, Duration::from_secs(120))
-            .await
-            .expect("SPV did not reach Running state within 120s");
-        tracing::info!("SPV fully synced — mempool bloom filter active");
-
         // Verify balance is above minimum threshold
         funding::verify_framework_funded(&app_context, framework_wallet_hash).await;
 

tests/backend-e2e/tx_is_ours.rs

@@ -37,6 +37,11 @@ async fn test_spv_transactions_is_ours_flag() {
         .await
         .expect("Wallet A should have spendable funds");
 
+    // Allow bloom filter to propagate to peers so B's addresses are
+    // monitored before we broadcast A→B. Without this, peers may not
+    // relay the tx back through B's filter.
+    tokio::time::sleep(Duration::from_secs(2)).await;
+
     // Send from A to B
     let request = WalletPaymentRequest {
         recipients: vec![PaymentRecipient {