Offload blocking wallet calls to spawn_blocking

BDK wallet operations (apply_update, create_funding_transaction,
get_spendable_amount_sats) internally acquire a mutex and walk
the checkpoint chain, giving O(T*C) runtime where T is the
number of transactions and C is the number of checkpoints.
Running these on a tokio worker thread blocks the entire async
executor during sync-heavy workloads.

Move these calls to spawn_blocking so they run on the tokio
blocking threadpool instead. This is a stopgap until the underlying
issue is addressed in BDK.

JoinError (panic or runtime shutdown) is mapped into the
existing error paths at each call site: WalletOperationFailed
for esplora sync and funding tx creation, unwrap_or(0) for
spendable balance checks (conservative default that skips
channel opens).

Author: amackillop

src/chain/bitcoind.rs

@@ -436,7 +436,17 @@ impl BitcoindChainSource {
 		let cur_height = channel_manager.current_best_block().height;
 
 		let now = SystemTime::now();
-		let bdk_unconfirmed_txids = onchain_wallet.get_unconfirmed_txids();
+
+		let wallet_ref = onchain_wallet.clone();
+		let bdk_unconfirmed_txids = tokio::task::spawn_blocking(move || {
+			wallet_ref.get_unconfirmed_txids()
+		})
+		.await
+		.map_err(|e| {
+			log_error!(self.logger, "Failed to retrieve unconfirmed txids: {}", e);
+			Error::WalletOperationFailed
+		})?;
+
 		match self
 			.api_client
 			.get_updated_mempool_transactions(cur_height, bdk_unconfirmed_txids)
@@ -450,11 +460,18 @@ impl BitcoindChainSource {
 					evicted_txids.len(),
 					now.elapsed().unwrap().as_millis()
 				);
-				onchain_wallet.apply_mempool_txs(unconfirmed_txs, evicted_txids).unwrap_or_else(
-					|e| {
-						log_error!(self.logger, "Failed to apply mempool transactions: {:?}", e);
-					},
-				);
+
+				let apply_res = tokio::task::spawn_blocking(move || {
+					onchain_wallet.apply_mempool_txs(unconfirmed_txs, evicted_txids)
+				})
+				.await
+				.map_err(|e| {
+					log_error!(self.logger, "Applying mempool transactions panicked: {}", e);
+					Error::WalletOperationFailed
+				})?;
+				apply_res.unwrap_or_else(|e| {
+					log_error!(self.logger, "Failed to apply mempool transactions: {:?}", e);
+				});
 			},
 			Err(e) => {
 				log_error!(self.logger, "Failed to poll for mempool transactions: {:?}", e);

src/chain/esplora.rs

@@ -111,7 +111,15 @@ impl EsploraChainSource {
 				let now = Instant::now();
 				match $sync_future.await {
 					Ok(res) => match res {
-						Ok(update) => match onchain_wallet.apply_update(update) {
+						Ok(update) => {
+						let w = Arc::clone(&onchain_wallet);
+						let apply_res = tokio::task::spawn_blocking(move || w.apply_update(update))
+							.await
+							.map_err(|e| {
+								log_error!(self.logger, "Failed to apply wallet update: {}", e);
+								Error::WalletOperationFailed
+							})?;
+						match apply_res {
 							Ok(()) => {
 								log_info!(
 									self.logger,
@@ -135,6 +143,7 @@ impl EsploraChainSource {
 									Ok(())
 							},
 							Err(e) => Err(e),
+						}
 						},
 						Err(e) => match *e {
 							esplora_client::Error::Reqwest(he) => {

src/event.rs

@@ -564,12 +564,18 @@ where
 
 				// Sign the final funding transaction and broadcast it.
 				let channel_amount = Amount::from_sat(channel_value_satoshis);
-				match self.wallet.create_funding_transaction(
-					output_script,
-					channel_amount,
-					confirmation_target,
-					locktime,
-				) {
+				let w = Arc::clone(&self.wallet);
+				let funding_res = tokio::task::spawn_blocking(move || {
+					w.create_funding_transaction(
+						output_script, channel_amount, confirmation_target, locktime,
+					)
+				})
+				.await
+				.unwrap_or_else(|e| {
+					log_error!(self.logger, "Failed to create funding transaction: {}", e);
+					Err(Error::WalletOperationFailed)
+				});
+				match funding_res {
 					Ok(final_tx) => {
 						let needs_manual_broadcast =
 							self.liquidity_source.as_ref().map_or(false, |ls| {

src/liquidity.rs

@@ -879,8 +879,16 @@ where
 				let channel_amount_sats = (amt_to_forward_msat + over_provisioning_msat) / 1000;
 				let cur_anchor_reserve_sats =
 					total_anchor_channels_reserve_sats(&self.channel_manager, &self.config);
-				let spendable_amount_sats =
-					self.wallet.get_spendable_amount_sats(cur_anchor_reserve_sats).unwrap_or(0);
+				let w = Arc::clone(&self.wallet);
+				let spendable_amount_sats = tokio::task::spawn_blocking(move || {
+					w.get_spendable_amount_sats(cur_anchor_reserve_sats)
+				})
+				.await
+				.unwrap_or_else(|e| {
+					log_error!(self.logger, "Failed to get spendable amount: {}", e);
+					Err(Error::WalletOperationFailed)
+				})
+				.unwrap_or(0);
 				let required_funds_sats = channel_amount_sats
 					+ self.config.anchor_channels_config.as_ref().map_or(0, |c| {
 						if init_features.requires_anchors_zero_fee_htlc_tx()
@@ -1242,8 +1250,16 @@ where
 				}
 				let cur_anchor_reserve_sats =
 					total_anchor_channels_reserve_sats(&self.channel_manager, &self.config);
-				let spendable_amount_sats =
-					self.wallet.get_spendable_amount_sats(cur_anchor_reserve_sats).unwrap_or(0);
+				let w = Arc::clone(&self.wallet);
+				let spendable_amount_sats = tokio::task::spawn_blocking(move || {
+					w.get_spendable_amount_sats(cur_anchor_reserve_sats)
+				})
+				.await
+				.unwrap_or_else(|e| {
+					log_error!(self.logger, "Failed to get spendable amount: {}", e);
+					Err(Error::WalletOperationFailed)
+				})
+				.unwrap_or(0);
 				let required_funds_sats = channel_amount_sats
 					+ self.config.anchor_channels_config.as_ref().map_or(0, |c| {
 						if init_features.requires_anchors_zero_fee_htlc_tx()