Avoid wallet persistence deadlocks

src/chain/bitcoind.rs

@@ -461,11 +461,12 @@ impl BitcoindChainSource {
  evicted_txids.len(),
  elapsed_ms,
  );
- onchain_wallet.apply_mempool_txs(unconfirmed_txs, evicted_txids).unwrap_or_else(
- |e| {
+ onchain_wallet
+ .apply_mempool_txs(unconfirmed_txs, evicted_txids)
+ .await
+ .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/electrum.rs

@@ -182,7 +182,7 @@ impl ElectrumChainSource {
  update_res: Result<BdkUpdate, Error>, now: Instant,
 	) -> Result<(), Error> {
  match update_res {
- Ok(update) => match onchain_wallet.apply_update(update) {
+ Ok(update) => match onchain_wallet.apply_update(update).await {
  Ok(()) => {
  log_debug!(
  self.logger,

src/chain/esplora.rs

@@ -146,7 +146,7 @@ impl EsploraChainSource {
  let now = Instant::now();
  match $sync_future.await {
  Ok(res) => match res {
- Ok(update) => match onchain_wallet.apply_update(update) {
+ Ok(update) => match onchain_wallet.apply_update(update).await {
  Ok(()) => {
  log_debug!(
  self.logger,

src/event.rs

@@ -694,12 +694,16 @@ 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 funding_transaction = self
+ .wallet
+ .create_funding_transaction(
+ output_script,
+ channel_amount,
+ confirmation_target,
+ locktime,
+ )
+ .await;
+ match funding_transaction {
  Ok(final_tx) => {
  let needs_manual_broadcast = self
  .liquidity_source
@@ -1743,7 +1747,7 @@ where
  })
  .collect(),
  };
- if let Err(e) = self.wallet.cancel_tx(tx) {
+ if let Err(e) = self.wallet.cancel_tx(tx).await {
  log_error!(self.logger, "Failed reclaiming unused addresses: {}", e);
  return Err(ReplayEvent());
  }

src/lib.rs

@@ -1106,6 +1106,7 @@ impl Node {
  Arc::clone(&self.channel_manager),
  Arc::clone(&self.config),
  Arc::clone(&self.is_running),
+ Arc::clone(&self.runtime),
  Arc::clone(&self.logger),
  )
 	}
@@ -1118,6 +1119,7 @@ impl Node {
  Arc::clone(&self.channel_manager),
  Arc::clone(&self.config),
  Arc::clone(&self.is_running),
+ Arc::clone(&self.runtime),
  Arc::clone(&self.logger),
  ))
 	}

src/liquidity/client/lsps1.rs

@@ -488,7 +488,7 @@ impl LSPS1Liquidity {
 
  log_info!(self.logger, "Connected to LSP {}@{}. ", lsps1_node.node_id, lsps1_node.address);
 
- let refund_address = self.wallet.get_new_address()?;
+ let refund_address = self.runtime.block_on(self.wallet.get_new_address())?;
 
  let liquidity_source = Arc::clone(&self.liquidity_source);
  let response = self.runtime.block_on(async move {

src/payment/onchain.rs

@@ -15,6 +15,7 @@ use lightning::ln::channelmanager::PaymentId;
 use crate::config::Config;
 use crate::error::Error;
 use crate::logger::{log_info, LdkLogger, Logger};
+use crate::runtime::Runtime;
 use crate::types::{ChannelManager, Wallet};
 use crate::wallet::OnchainSendAmount;
 
@@ -47,23 +48,38 @@ pub struct OnchainPayment {
 	channel_manager: Arc<ChannelManager>,
 	config: Arc<Config>,
 	is_running: Arc<RwLock<bool>>,
+	runtime: Arc<Runtime>,
 	logger: Arc<Logger>,
 }
 
 impl OnchainPayment {
 	pub(crate) fn new(
  wallet: Arc<Wallet>, channel_manager: Arc<ChannelManager>, config: Arc<Config>,
- is_running: Arc<RwLock<bool>>, logger: Arc<Logger>,
+ is_running: Arc<RwLock<bool>>, runtime: Arc<Runtime>, logger: Arc<Logger>,
 	) -> Self {
- Self { wallet, channel_manager, config, is_running, logger }
+ Self { wallet, channel_manager, config, is_running, runtime, logger }
+	}
+
+	pub(crate) async fn send_to_address_inner(
+ &self, address: &bitcoin::Address, amount_sats: u64, fee_rate: Option<bitcoin::FeeRate>,
+	) -> Result<Txid, Error> {
+ if !*self.is_running.read().expect("lock") {
+ return Err(Error::NotRunning);
+ }
+
+ let cur_anchor_reserve_sats =
+ crate::total_anchor_channels_reserve_sats(&self.channel_manager, &self.config);
+ let send_amount =
+ OnchainSendAmount::ExactRetainingReserve { amount_sats, cur_anchor_reserve_sats };
+ self.wallet.send_to_address(address, send_amount, fee_rate).await
 	}
 }
 
 #[cfg_attr(feature = "uniffi", uniffi::export)]
 impl OnchainPayment {
 	/// Retrieve a new on-chain/funding address.
 	pub fn new_address(&self) -> Result<Address, Error> {
- let funding_address = self.wallet.get_new_address()?;
+ let funding_address = self.runtime.block_on(self.wallet.get_new_address())?;
  log_info!(self.logger, "Generated new funding address: {}", funding_address);
  Ok(funding_address)
 	}
@@ -80,16 +96,8 @@ impl OnchainPayment {
 	pub fn send_to_address(
  &self, address: &bitcoin::Address, amount_sats: u64, fee_rate: Option<FeeRate>,
 	) -> Result<Txid, Error> {
- if !*self.is_running.read().expect("lock") {
- return Err(Error::NotRunning);
- }
-
- let cur_anchor_reserve_sats =
- crate::total_anchor_channels_reserve_sats(&self.channel_manager, &self.config);
- let send_amount =
- OnchainSendAmount::ExactRetainingReserve { amount_sats, cur_anchor_reserve_sats };
  let fee_rate_opt = maybe_map_fee_rate_opt!(fee_rate);
- self.wallet.send_to_address(address, send_amount, fee_rate_opt)
+ self.runtime.block_on(self.send_to_address_inner(address, amount_sats, fee_rate_opt))
 	}
 
 	/// Send an on-chain payment to the given address, draining the available funds.
@@ -123,7 +131,7 @@ impl OnchainPayment {
  };
 
  let fee_rate_opt = maybe_map_fee_rate_opt!(fee_rate);
- self.wallet.send_to_address(address, send_amount, fee_rate_opt)
+ self.runtime.block_on(self.wallet.send_to_address(address, send_amount, fee_rate_opt))
 	}
 
 	/// Attempt to bump the fee of an unconfirmed transaction using Replace-by-Fee (RBF).
@@ -146,6 +154,10 @@ impl OnchainPayment {
  let cur_anchor_reserve_sats =
  crate::total_anchor_channels_reserve_sats(&self.channel_manager, &self.config);
  let fee_rate_opt = maybe_map_fee_rate_opt!(fee_rate);
- self.wallet.bump_fee_rbf(payment_id, fee_rate_opt, cur_anchor_reserve_sats)
+ self.runtime.block_on(self.wallet.bump_fee_rbf(
+ payment_id,
+ fee_rate_opt,
+ cur_anchor_reserve_sats,
+ ))
 	}
 }

src/payment/unified.rs

@@ -328,7 +328,10 @@ impl UnifiedPayment {
  Error::InvalidAmount
  })?;
 
- let txid = self.onchain_payment.send_to_address(&address, amt_sats, None)?;
+ let txid = self
+ .onchain_payment
+ .send_to_address_inner(&address, amt_sats, None)
+ .await?;
  return Ok(UnifiedPaymentResult::Onchain { txid });
  },
  }

src/chain/bitcoind.rs

@@ -461,11 +461,12 @@ impl BitcoindChainSource {
  evicted_txids.len(),
  elapsed_ms,
  );
- onchain_wallet.apply_mempool_txs(unconfirmed_txs, evicted_txids).unwrap_or_else(
- |e| {
+ onchain_wallet
+ .apply_mempool_txs(unconfirmed_txs, evicted_txids)
+ .await
+ .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/electrum.rs

@@ -182,7 +182,7 @@ impl ElectrumChainSource {
  update_res: Result<BdkUpdate, Error>, now: Instant,
 	) -> Result<(), Error> {
  match update_res {
- Ok(update) => match onchain_wallet.apply_update(update) {
+ Ok(update) => match onchain_wallet.apply_update(update).await {
  Ok(()) => {
  log_debug!(
  self.logger,

src/chain/esplora.rs

@@ -146,7 +146,7 @@ impl EsploraChainSource {
  let now = Instant::now();
  match $sync_future.await {
  Ok(res) => match res {
- Ok(update) => match onchain_wallet.apply_update(update) {
+ Ok(update) => match onchain_wallet.apply_update(update).await {
  Ok(()) => {
  log_debug!(
  self.logger,

src/event.rs

@@ -694,12 +694,16 @@ 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 funding_transaction = self
+ .wallet
+ .create_funding_transaction(
+ output_script,
+ channel_amount,
+ confirmation_target,
+ locktime,
+ )
+ .await;
+ match funding_transaction {
  Ok(final_tx) => {
  let needs_manual_broadcast = self
  .liquidity_source
@@ -1743,7 +1747,7 @@ where
  })
  .collect(),
  };
- if let Err(e) = self.wallet.cancel_tx(tx) {
+ if let Err(e) = self.wallet.cancel_tx(tx).await {
  log_error!(self.logger, "Failed reclaiming unused addresses: {}", e);
  return Err(ReplayEvent());
  }

src/lib.rs

@@ -1106,6 +1106,7 @@ impl Node {
  Arc::clone(&self.channel_manager),
  Arc::clone(&self.config),
  Arc::clone(&self.is_running),
+ Arc::clone(&self.runtime),
  Arc::clone(&self.logger),
  )
 	}
@@ -1118,6 +1119,7 @@ impl Node {
  Arc::clone(&self.channel_manager),
  Arc::clone(&self.config),
  Arc::clone(&self.is_running),
+ Arc::clone(&self.runtime),
  Arc::clone(&self.logger),
  ))
 	}

src/liquidity/client/lsps1.rs

@@ -488,7 +488,7 @@ impl LSPS1Liquidity {
 
  log_info!(self.logger, "Connected to LSP {}@{}. ", lsps1_node.node_id, lsps1_node.address);
 
- let refund_address = self.wallet.get_new_address()?;
+ let refund_address = self.runtime.block_on(self.wallet.get_new_address())?;
 
  let liquidity_source = Arc::clone(&self.liquidity_source);
  let response = self.runtime.block_on(async move {

src/payment/onchain.rs

@@ -15,6 +15,7 @@ use lightning::ln::channelmanager::PaymentId;
 use crate::config::Config;
 use crate::error::Error;
 use crate::logger::{log_info, LdkLogger, Logger};
+use crate::runtime::Runtime;
 use crate::types::{ChannelManager, Wallet};
 use crate::wallet::OnchainSendAmount;
 
@@ -47,23 +48,38 @@ pub struct OnchainPayment {
 	channel_manager: Arc<ChannelManager>,
 	config: Arc<Config>,
 	is_running: Arc<RwLock<bool>>,
+	runtime: Arc<Runtime>,
 	logger: Arc<Logger>,
 }
 
 impl OnchainPayment {
 	pub(crate) fn new(
  wallet: Arc<Wallet>, channel_manager: Arc<ChannelManager>, config: Arc<Config>,
- is_running: Arc<RwLock<bool>>, logger: Arc<Logger>,
+ is_running: Arc<RwLock<bool>>, runtime: Arc<Runtime>, logger: Arc<Logger>,
 	) -> Self {
- Self { wallet, channel_manager, config, is_running, logger }
+ Self { wallet, channel_manager, config, is_running, runtime, logger }
+	}
+
+	pub(crate) async fn send_to_address_inner(
+ &self, address: &bitcoin::Address, amount_sats: u64, fee_rate: Option<bitcoin::FeeRate>,
+	) -> Result<Txid, Error> {
+ if !*self.is_running.read().expect("lock") {
+ return Err(Error::NotRunning);
+ }
+
+ let cur_anchor_reserve_sats =
+ crate::total_anchor_channels_reserve_sats(&self.channel_manager, &self.config);
+ let send_amount =
+ OnchainSendAmount::ExactRetainingReserve { amount_sats, cur_anchor_reserve_sats };
+ self.wallet.send_to_address(address, send_amount, fee_rate).await
 	}
 }
 
 #[cfg_attr(feature = "uniffi", uniffi::export)]
 impl OnchainPayment {
 	/// Retrieve a new on-chain/funding address.
 	pub fn new_address(&self) -> Result<Address, Error> {
- let funding_address = self.wallet.get_new_address()?;
+ let funding_address = self.runtime.block_on(self.wallet.get_new_address())?;
  log_info!(self.logger, "Generated new funding address: {}", funding_address);
  Ok(funding_address)
 	}
@@ -80,16 +96,8 @@ impl OnchainPayment {
 	pub fn send_to_address(
  &self, address: &bitcoin::Address, amount_sats: u64, fee_rate: Option<FeeRate>,
 	) -> Result<Txid, Error> {
- if !*self.is_running.read().expect("lock") {
- return Err(Error::NotRunning);
- }
-
- let cur_anchor_reserve_sats =
- crate::total_anchor_channels_reserve_sats(&self.channel_manager, &self.config);
- let send_amount =
- OnchainSendAmount::ExactRetainingReserve { amount_sats, cur_anchor_reserve_sats };
  let fee_rate_opt = maybe_map_fee_rate_opt!(fee_rate);
- self.wallet.send_to_address(address, send_amount, fee_rate_opt)
+ self.runtime.block_on(self.send_to_address_inner(address, amount_sats, fee_rate_opt))
 	}
 
 	/// Send an on-chain payment to the given address, draining the available funds.
@@ -123,7 +131,7 @@ impl OnchainPayment {
  };
 
  let fee_rate_opt = maybe_map_fee_rate_opt!(fee_rate);
- self.wallet.send_to_address(address, send_amount, fee_rate_opt)
+ self.runtime.block_on(self.wallet.send_to_address(address, send_amount, fee_rate_opt))
 	}
 
 	/// Attempt to bump the fee of an unconfirmed transaction using Replace-by-Fee (RBF).
@@ -146,6 +154,10 @@ impl OnchainPayment {
  let cur_anchor_reserve_sats =
  crate::total_anchor_channels_reserve_sats(&self.channel_manager, &self.config);
  let fee_rate_opt = maybe_map_fee_rate_opt!(fee_rate);
- self.wallet.bump_fee_rbf(payment_id, fee_rate_opt, cur_anchor_reserve_sats)
+ self.runtime.block_on(self.wallet.bump_fee_rbf(
+ payment_id,
+ fee_rate_opt,
+ cur_anchor_reserve_sats,
+ ))
 	}
 }

src/payment/unified.rs

@@ -328,7 +328,10 @@ impl UnifiedPayment {
  Error::InvalidAmount
  })?;
 
- let txid = self.onchain_payment.send_to_address(&address, amt_sats, None)?;
+ let txid = self
+ .onchain_payment
+ .send_to_address_inner(&address, amt_sats, None)
+ .await?;
  return Ok(UnifiedPaymentResult::Onchain { txid });
  },
  }

src/chain/bitcoind.rs

@@ -461,11 +461,12 @@ impl BitcoindChainSource {
  evicted_txids.len(),
  elapsed_ms,
  );
- onchain_wallet.apply_mempool_txs(unconfirmed_txs, evicted_txids).unwrap_or_else(
- |e| {
+ onchain_wallet
+ .apply_mempool_txs(unconfirmed_txs, evicted_txids)
+ .await
+ .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/electrum.rs

@@ -182,7 +182,7 @@ impl ElectrumChainSource {
  update_res: Result<BdkUpdate, Error>, now: Instant,
 	) -> Result<(), Error> {
  match update_res {
- Ok(update) => match onchain_wallet.apply_update(update) {
+ Ok(update) => match onchain_wallet.apply_update(update).await {
  Ok(()) => {
  log_debug!(
  self.logger,

src/chain/esplora.rs

@@ -146,7 +146,7 @@ impl EsploraChainSource {
  let now = Instant::now();
  match $sync_future.await {
  Ok(res) => match res {
- Ok(update) => match onchain_wallet.apply_update(update) {
+ Ok(update) => match onchain_wallet.apply_update(update).await {
  Ok(()) => {
  log_debug!(
  self.logger,

src/event.rs

@@ -694,12 +694,16 @@ 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 funding_transaction = self
+ .wallet
+ .create_funding_transaction(
+ output_script,
+ channel_amount,
+ confirmation_target,
+ locktime,
+ )
+ .await;
+ match funding_transaction {
  Ok(final_tx) => {
  let needs_manual_broadcast = self
  .liquidity_source
@@ -1743,7 +1747,7 @@ where
  })
  .collect(),
  };
- if let Err(e) = self.wallet.cancel_tx(tx) {
+ if let Err(e) = self.wallet.cancel_tx(tx).await {
  log_error!(self.logger, "Failed reclaiming unused addresses: {}", e);
  return Err(ReplayEvent());
  }

src/lib.rs

@@ -1106,6 +1106,7 @@ impl Node {
  Arc::clone(&self.channel_manager),
  Arc::clone(&self.config),
  Arc::clone(&self.is_running),
+ Arc::clone(&self.runtime),
  Arc::clone(&self.logger),
  )
 	}
@@ -1118,6 +1119,7 @@ impl Node {
  Arc::clone(&self.channel_manager),
  Arc::clone(&self.config),
  Arc::clone(&self.is_running),
+ Arc::clone(&self.runtime),
  Arc::clone(&self.logger),
  ))
 	}

src/liquidity/client/lsps1.rs

@@ -488,7 +488,7 @@ impl LSPS1Liquidity {
 
  log_info!(self.logger, "Connected to LSP {}@{}. ", lsps1_node.node_id, lsps1_node.address);
 
- let refund_address = self.wallet.get_new_address()?;
+ let refund_address = self.runtime.block_on(self.wallet.get_new_address())?;
 
  let liquidity_source = Arc::clone(&self.liquidity_source);
  let response = self.runtime.block_on(async move {

src/payment/onchain.rs

@@ -15,6 +15,7 @@ use lightning::ln::channelmanager::PaymentId;
 use crate::config::Config;
 use crate::error::Error;
 use crate::logger::{log_info, LdkLogger, Logger};
+use crate::runtime::Runtime;
 use crate::types::{ChannelManager, Wallet};
 use crate::wallet::OnchainSendAmount;
 
@@ -47,23 +48,38 @@ pub struct OnchainPayment {
 	channel_manager: Arc<ChannelManager>,
 	config: Arc<Config>,
 	is_running: Arc<RwLock<bool>>,
+	runtime: Arc<Runtime>,
 	logger: Arc<Logger>,
 }
 
 impl OnchainPayment {
 	pub(crate) fn new(
  wallet: Arc<Wallet>, channel_manager: Arc<ChannelManager>, config: Arc<Config>,
- is_running: Arc<RwLock<bool>>, logger: Arc<Logger>,
+ is_running: Arc<RwLock<bool>>, runtime: Arc<Runtime>, logger: Arc<Logger>,
 	) -> Self {
- Self { wallet, channel_manager, config, is_running, logger }
+ Self { wallet, channel_manager, config, is_running, runtime, logger }
+	}
+
+	pub(crate) async fn send_to_address_inner(
+ &self, address: &bitcoin::Address, amount_sats: u64, fee_rate: Option<bitcoin::FeeRate>,
+	) -> Result<Txid, Error> {
+ if !*self.is_running.read().expect("lock") {
+ return Err(Error::NotRunning);
+ }
+
+ let cur_anchor_reserve_sats =
+ crate::total_anchor_channels_reserve_sats(&self.channel_manager, &self.config);
+ let send_amount =
+ OnchainSendAmount::ExactRetainingReserve { amount_sats, cur_anchor_reserve_sats };
+ self.wallet.send_to_address(address, send_amount, fee_rate).await
 	}
 }
 
 #[cfg_attr(feature = "uniffi", uniffi::export)]
 impl OnchainPayment {
 	/// Retrieve a new on-chain/funding address.
 	pub fn new_address(&self) -> Result<Address, Error> {
- let funding_address = self.wallet.get_new_address()?;
+ let funding_address = self.runtime.block_on(self.wallet.get_new_address())?;
  log_info!(self.logger, "Generated new funding address: {}", funding_address);
  Ok(funding_address)
 	}
@@ -80,16 +96,8 @@ impl OnchainPayment {
 	pub fn send_to_address(
  &self, address: &bitcoin::Address, amount_sats: u64, fee_rate: Option<FeeRate>,
 	) -> Result<Txid, Error> {
- if !*self.is_running.read().expect("lock") {
- return Err(Error::NotRunning);
- }
-
- let cur_anchor_reserve_sats =
- crate::total_anchor_channels_reserve_sats(&self.channel_manager, &self.config);
- let send_amount =
- OnchainSendAmount::ExactRetainingReserve { amount_sats, cur_anchor_reserve_sats };
  let fee_rate_opt = maybe_map_fee_rate_opt!(fee_rate);
- self.wallet.send_to_address(address, send_amount, fee_rate_opt)
+ self.runtime.block_on(self.send_to_address_inner(address, amount_sats, fee_rate_opt))
 	}
 
 	/// Send an on-chain payment to the given address, draining the available funds.
@@ -123,7 +131,7 @@ impl OnchainPayment {
  };
 
  let fee_rate_opt = maybe_map_fee_rate_opt!(fee_rate);
- self.wallet.send_to_address(address, send_amount, fee_rate_opt)
+ self.runtime.block_on(self.wallet.send_to_address(address, send_amount, fee_rate_opt))
 	}
 
 	/// Attempt to bump the fee of an unconfirmed transaction using Replace-by-Fee (RBF).
@@ -146,6 +154,10 @@ impl OnchainPayment {
  let cur_anchor_reserve_sats =
  crate::total_anchor_channels_reserve_sats(&self.channel_manager, &self.config);
  let fee_rate_opt = maybe_map_fee_rate_opt!(fee_rate);
- self.wallet.bump_fee_rbf(payment_id, fee_rate_opt, cur_anchor_reserve_sats)
+ self.runtime.block_on(self.wallet.bump_fee_rbf(
+ payment_id,
+ fee_rate_opt,
+ cur_anchor_reserve_sats,
+ ))
 	}
 }

src/payment/unified.rs

@@ -328,7 +328,10 @@ impl UnifiedPayment {
  Error::InvalidAmount
  })?;
 
- let txid = self.onchain_payment.send_to_address(&address, amt_sats, None)?;
+ let txid = self
+ .onchain_payment
+ .send_to_address_inner(&address, amt_sats, None)
+ .await?;
  return Ok(UnifiedPaymentResult::Onchain { txid });
  },
  }

src/chain/bitcoind.rs

@@ -461,11 +461,12 @@ impl BitcoindChainSource {
  evicted_txids.len(),
  elapsed_ms,
  );
- onchain_wallet.apply_mempool_txs(unconfirmed_txs, evicted_txids).unwrap_or_else(
- |e| {
+ onchain_wallet
+ .apply_mempool_txs(unconfirmed_txs, evicted_txids)
+ .await
+ .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/electrum.rs

@@ -182,7 +182,7 @@ impl ElectrumChainSource {
  update_res: Result<BdkUpdate, Error>, now: Instant,
 	) -> Result<(), Error> {
  match update_res {
- Ok(update) => match onchain_wallet.apply_update(update) {
+ Ok(update) => match onchain_wallet.apply_update(update).await {
  Ok(()) => {
  log_debug!(
  self.logger,

src/chain/esplora.rs

@@ -146,7 +146,7 @@ impl EsploraChainSource {
  let now = Instant::now();
  match $sync_future.await {
  Ok(res) => match res {
- Ok(update) => match onchain_wallet.apply_update(update) {
+ Ok(update) => match onchain_wallet.apply_update(update).await {
  Ok(()) => {
  log_debug!(
  self.logger,

src/event.rs

@@ -694,12 +694,16 @@ 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 funding_transaction = self
+ .wallet
+ .create_funding_transaction(
+ output_script,
+ channel_amount,
+ confirmation_target,
+ locktime,
+ )
+ .await;
+ match funding_transaction {
  Ok(final_tx) => {
  let needs_manual_broadcast = self
  .liquidity_source
@@ -1743,7 +1747,7 @@ where
  })
  .collect(),
  };
- if let Err(e) = self.wallet.cancel_tx(tx) {
+ if let Err(e) = self.wallet.cancel_tx(tx).await {
  log_error!(self.logger, "Failed reclaiming unused addresses: {}", e);
  return Err(ReplayEvent());
  }

src/lib.rs

@@ -1106,6 +1106,7 @@ impl Node {
  Arc::clone(&self.channel_manager),
  Arc::clone(&self.config),
  Arc::clone(&self.is_running),
+ Arc::clone(&self.runtime),
  Arc::clone(&self.logger),
  )
 	}
@@ -1118,6 +1119,7 @@ impl Node {
  Arc::clone(&self.channel_manager),
  Arc::clone(&self.config),
  Arc::clone(&self.is_running),
+ Arc::clone(&self.runtime),
  Arc::clone(&self.logger),
  ))
 	}

src/liquidity/client/lsps1.rs

@@ -488,7 +488,7 @@ impl LSPS1Liquidity {
 
  log_info!(self.logger, "Connected to LSP {}@{}. ", lsps1_node.node_id, lsps1_node.address);
 
- let refund_address = self.wallet.get_new_address()?;
+ let refund_address = self.runtime.block_on(self.wallet.get_new_address())?;
 
  let liquidity_source = Arc::clone(&self.liquidity_source);
  let response = self.runtime.block_on(async move {

src/payment/onchain.rs

@@ -15,6 +15,7 @@ use lightning::ln::channelmanager::PaymentId;
 use crate::config::Config;
 use crate::error::Error;
 use crate::logger::{log_info, LdkLogger, Logger};
+use crate::runtime::Runtime;
 use crate::types::{ChannelManager, Wallet};
 use crate::wallet::OnchainSendAmount;
 
@@ -47,23 +48,38 @@ pub struct OnchainPayment {
 	channel_manager: Arc<ChannelManager>,
 	config: Arc<Config>,
 	is_running: Arc<RwLock<bool>>,
+	runtime: Arc<Runtime>,
 	logger: Arc<Logger>,
 }
 
 impl OnchainPayment {
 	pub(crate) fn new(
  wallet: Arc<Wallet>, channel_manager: Arc<ChannelManager>, config: Arc<Config>,
- is_running: Arc<RwLock<bool>>, logger: Arc<Logger>,
+ is_running: Arc<RwLock<bool>>, runtime: Arc<Runtime>, logger: Arc<Logger>,
 	) -> Self {
- Self { wallet, channel_manager, config, is_running, logger }
+ Self { wallet, channel_manager, config, is_running, runtime, logger }
+	}
+
+	pub(crate) async fn send_to_address_inner(
+ &self, address: &bitcoin::Address, amount_sats: u64, fee_rate: Option<bitcoin::FeeRate>,
+	) -> Result<Txid, Error> {
+ if !*self.is_running.read().expect("lock") {
+ return Err(Error::NotRunning);
+ }
+
+ let cur_anchor_reserve_sats =
+ crate::total_anchor_channels_reserve_sats(&self.channel_manager, &self.config);
+ let send_amount =
+ OnchainSendAmount::ExactRetainingReserve { amount_sats, cur_anchor_reserve_sats };
+ self.wallet.send_to_address(address, send_amount, fee_rate).await
 	}
 }
 
 #[cfg_attr(feature = "uniffi", uniffi::export)]
 impl OnchainPayment {
 	/// Retrieve a new on-chain/funding address.
 	pub fn new_address(&self) -> Result<Address, Error> {
- let funding_address = self.wallet.get_new_address()?;
+ let funding_address = self.runtime.block_on(self.wallet.get_new_address())?;
  log_info!(self.logger, "Generated new funding address: {}", funding_address);
  Ok(funding_address)
 	}
@@ -80,16 +96,8 @@ impl OnchainPayment {
 	pub fn send_to_address(
  &self, address: &bitcoin::Address, amount_sats: u64, fee_rate: Option<FeeRate>,
 	) -> Result<Txid, Error> {
- if !*self.is_running.read().expect("lock") {
- return Err(Error::NotRunning);
- }
-
- let cur_anchor_reserve_sats =
- crate::total_anchor_channels_reserve_sats(&self.channel_manager, &self.config);
- let send_amount =
- OnchainSendAmount::ExactRetainingReserve { amount_sats, cur_anchor_reserve_sats };
  let fee_rate_opt = maybe_map_fee_rate_opt!(fee_rate);
- self.wallet.send_to_address(address, send_amount, fee_rate_opt)
+ self.runtime.block_on(self.send_to_address_inner(address, amount_sats, fee_rate_opt))
 	}
 
 	/// Send an on-chain payment to the given address, draining the available funds.
@@ -123,7 +131,7 @@ impl OnchainPayment {
  };
 
  let fee_rate_opt = maybe_map_fee_rate_opt!(fee_rate);
- self.wallet.send_to_address(address, send_amount, fee_rate_opt)
+ self.runtime.block_on(self.wallet.send_to_address(address, send_amount, fee_rate_opt))
 	}
 
 	/// Attempt to bump the fee of an unconfirmed transaction using Replace-by-Fee (RBF).
@@ -146,6 +154,10 @@ impl OnchainPayment {
  let cur_anchor_reserve_sats =
  crate::total_anchor_channels_reserve_sats(&self.channel_manager, &self.config);
  let fee_rate_opt = maybe_map_fee_rate_opt!(fee_rate);
- self.wallet.bump_fee_rbf(payment_id, fee_rate_opt, cur_anchor_reserve_sats)
+ self.runtime.block_on(self.wallet.bump_fee_rbf(
+ payment_id,
+ fee_rate_opt,
+ cur_anchor_reserve_sats,
+ ))
 	}
 }

src/payment/unified.rs

@@ -328,7 +328,10 @@ impl UnifiedPayment {
  Error::InvalidAmount
  })?;
 
- let txid = self.onchain_payment.send_to_address(&address, amt_sats, None)?;
+ let txid = self
+ .onchain_payment
+ .send_to_address_inner(&address, amt_sats, None)
+ .await?;
  return Ok(UnifiedPaymentResult::Onchain { txid });
  },
  }

src/chain/bitcoind.rs

@@ -461,11 +461,12 @@ impl BitcoindChainSource {
  evicted_txids.len(),
  elapsed_ms,
  );
- onchain_wallet.apply_mempool_txs(unconfirmed_txs, evicted_txids).unwrap_or_else(
- |e| {
+ onchain_wallet
+ .apply_mempool_txs(unconfirmed_txs, evicted_txids)
+ .await
+ .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/electrum.rs

@@ -182,7 +182,7 @@ impl ElectrumChainSource {
  update_res: Result<BdkUpdate, Error>, now: Instant,
 	) -> Result<(), Error> {
  match update_res {
- Ok(update) => match onchain_wallet.apply_update(update) {
+ Ok(update) => match onchain_wallet.apply_update(update).await {
  Ok(()) => {
  log_debug!(
  self.logger,

src/chain/esplora.rs

@@ -146,7 +146,7 @@ impl EsploraChainSource {
  let now = Instant::now();
  match $sync_future.await {
  Ok(res) => match res {
- Ok(update) => match onchain_wallet.apply_update(update) {
+ Ok(update) => match onchain_wallet.apply_update(update).await {
  Ok(()) => {
  log_debug!(
  self.logger,

src/event.rs

@@ -694,12 +694,16 @@ 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 funding_transaction = self
+ .wallet
+ .create_funding_transaction(
+ output_script,
+ channel_amount,
+ confirmation_target,
+ locktime,
+ )
+ .await;
+ match funding_transaction {
  Ok(final_tx) => {
  let needs_manual_broadcast = self
  .liquidity_source
@@ -1743,7 +1747,7 @@ where
  })
  .collect(),
  };
- if let Err(e) = self.wallet.cancel_tx(tx) {
+ if let Err(e) = self.wallet.cancel_tx(tx).await {
  log_error!(self.logger, "Failed reclaiming unused addresses: {}", e);
  return Err(ReplayEvent());
  }

src/lib.rs

@@ -1106,6 +1106,7 @@ impl Node {
  Arc::clone(&self.channel_manager),
  Arc::clone(&self.config),
  Arc::clone(&self.is_running),
+ Arc::clone(&self.runtime),
  Arc::clone(&self.logger),
  )
 	}
@@ -1118,6 +1119,7 @@ impl Node {
  Arc::clone(&self.channel_manager),
  Arc::clone(&self.config),
  Arc::clone(&self.is_running),
+ Arc::clone(&self.runtime),
  Arc::clone(&self.logger),
  ))
 	}

src/liquidity/client/lsps1.rs

@@ -488,7 +488,7 @@ impl LSPS1Liquidity {
 
  log_info!(self.logger, "Connected to LSP {}@{}. ", lsps1_node.node_id, lsps1_node.address);
 
- let refund_address = self.wallet.get_new_address()?;
+ let refund_address = self.runtime.block_on(self.wallet.get_new_address())?;
 
  let liquidity_source = Arc::clone(&self.liquidity_source);
  let response = self.runtime.block_on(async move {

src/payment/onchain.rs

@@ -15,6 +15,7 @@ use lightning::ln::channelmanager::PaymentId;
 use crate::config::Config;
 use crate::error::Error;
 use crate::logger::{log_info, LdkLogger, Logger};
+use crate::runtime::Runtime;
 use crate::types::{ChannelManager, Wallet};
 use crate::wallet::OnchainSendAmount;
 
@@ -47,23 +48,38 @@ pub struct OnchainPayment {
 	channel_manager: Arc<ChannelManager>,
 	config: Arc<Config>,
 	is_running: Arc<RwLock<bool>>,
+	runtime: Arc<Runtime>,
 	logger: Arc<Logger>,
 }
 
 impl OnchainPayment {
 	pub(crate) fn new(
  wallet: Arc<Wallet>, channel_manager: Arc<ChannelManager>, config: Arc<Config>,
- is_running: Arc<RwLock<bool>>, logger: Arc<Logger>,
+ is_running: Arc<RwLock<bool>>, runtime: Arc<Runtime>, logger: Arc<Logger>,
 	) -> Self {
- Self { wallet, channel_manager, config, is_running, logger }
+ Self { wallet, channel_manager, config, is_running, runtime, logger }
+	}
+
+	pub(crate) async fn send_to_address_inner(
+ &self, address: &bitcoin::Address, amount_sats: u64, fee_rate: Option<bitcoin::FeeRate>,
+	) -> Result<Txid, Error> {
+ if !*self.is_running.read().expect("lock") {
+ return Err(Error::NotRunning);
+ }
+
+ let cur_anchor_reserve_sats =
+ crate::total_anchor_channels_reserve_sats(&self.channel_manager, &self.config);
+ let send_amount =
+ OnchainSendAmount::ExactRetainingReserve { amount_sats, cur_anchor_reserve_sats };
+ self.wallet.send_to_address(address, send_amount, fee_rate).await
 	}
 }
 
 #[cfg_attr(feature = "uniffi", uniffi::export)]
 impl OnchainPayment {
 	/// Retrieve a new on-chain/funding address.
 	pub fn new_address(&self) -> Result<Address, Error> {
- let funding_address = self.wallet.get_new_address()?;
+ let funding_address = self.runtime.block_on(self.wallet.get_new_address())?;
  log_info!(self.logger, "Generated new funding address: {}", funding_address);
  Ok(funding_address)
 	}
@@ -80,16 +96,8 @@ impl OnchainPayment {
 	pub fn send_to_address(
  &self, address: &bitcoin::Address, amount_sats: u64, fee_rate: Option<FeeRate>,
 	) -> Result<Txid, Error> {
- if !*self.is_running.read().expect("lock") {
- return Err(Error::NotRunning);
- }
-
- let cur_anchor_reserve_sats =
- crate::total_anchor_channels_reserve_sats(&self.channel_manager, &self.config);
- let send_amount =
- OnchainSendAmount::ExactRetainingReserve { amount_sats, cur_anchor_reserve_sats };
  let fee_rate_opt = maybe_map_fee_rate_opt!(fee_rate);
- self.wallet.send_to_address(address, send_amount, fee_rate_opt)
+ self.runtime.block_on(self.send_to_address_inner(address, amount_sats, fee_rate_opt))
 	}
 
 	/// Send an on-chain payment to the given address, draining the available funds.
@@ -123,7 +131,7 @@ impl OnchainPayment {
  };
 
  let fee_rate_opt = maybe_map_fee_rate_opt!(fee_rate);
- self.wallet.send_to_address(address, send_amount, fee_rate_opt)
+ self.runtime.block_on(self.wallet.send_to_address(address, send_amount, fee_rate_opt))
 	}
 
 	/// Attempt to bump the fee of an unconfirmed transaction using Replace-by-Fee (RBF).
@@ -146,6 +154,10 @@ impl OnchainPayment {
  let cur_anchor_reserve_sats =
  crate::total_anchor_channels_reserve_sats(&self.channel_manager, &self.config);
  let fee_rate_opt = maybe_map_fee_rate_opt!(fee_rate);
- self.wallet.bump_fee_rbf(payment_id, fee_rate_opt, cur_anchor_reserve_sats)
+ self.runtime.block_on(self.wallet.bump_fee_rbf(
+ payment_id,
+ fee_rate_opt,
+ cur_anchor_reserve_sats,
+ ))
 	}
 }

src/payment/unified.rs

@@ -328,7 +328,10 @@ impl UnifiedPayment {
  Error::InvalidAmount
  })?;
 
- let txid = self.onchain_payment.send_to_address(&address, amt_sats, None)?;
+ let txid = self
+ .onchain_payment
+ .send_to_address_inner(&address, amt_sats, None)
+ .await?;
  return Ok(UnifiedPaymentResult::Onchain { txid });
  },
  }

src/chain/bitcoind.rs

@@ -461,11 +461,12 @@ impl BitcoindChainSource {
  evicted_txids.len(),
  elapsed_ms,
  );
- onchain_wallet.apply_mempool_txs(unconfirmed_txs, evicted_txids).unwrap_or_else(
- |e| {
+ onchain_wallet
+ .apply_mempool_txs(unconfirmed_txs, evicted_txids)
+ .await
+ .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/electrum.rs

@@ -182,7 +182,7 @@ impl ElectrumChainSource {
  update_res: Result<BdkUpdate, Error>, now: Instant,
 	) -> Result<(), Error> {
  match update_res {
- Ok(update) => match onchain_wallet.apply_update(update) {
+ Ok(update) => match onchain_wallet.apply_update(update).await {
  Ok(()) => {
  log_debug!(
  self.logger,

src/chain/esplora.rs

@@ -146,7 +146,7 @@ impl EsploraChainSource {
  let now = Instant::now();
  match $sync_future.await {
  Ok(res) => match res {
- Ok(update) => match onchain_wallet.apply_update(update) {
+ Ok(update) => match onchain_wallet.apply_update(update).await {
  Ok(()) => {
  log_debug!(
  self.logger,

src/event.rs

@@ -694,12 +694,16 @@ 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 funding_transaction = self
+ .wallet
+ .create_funding_transaction(
+ output_script,
+ channel_amount,
+ confirmation_target,
+ locktime,
+ )
+ .await;
+ match funding_transaction {
  Ok(final_tx) => {
  let needs_manual_broadcast = self
  .liquidity_source
@@ -1743,7 +1747,7 @@ where
  })
  .collect(),
  };
- if let Err(e) = self.wallet.cancel_tx(tx) {
+ if let Err(e) = self.wallet.cancel_tx(tx).await {
  log_error!(self.logger, "Failed reclaiming unused addresses: {}", e);
  return Err(ReplayEvent());
  }

src/lib.rs

@@ -1106,6 +1106,7 @@ impl Node {
  Arc::clone(&self.channel_manager),
  Arc::clone(&self.config),
  Arc::clone(&self.is_running),
+ Arc::clone(&self.runtime),
  Arc::clone(&self.logger),
  )
 	}
@@ -1118,6 +1119,7 @@ impl Node {
  Arc::clone(&self.channel_manager),
  Arc::clone(&self.config),
  Arc::clone(&self.is_running),
+ Arc::clone(&self.runtime),
  Arc::clone(&self.logger),
  ))
 	}

src/liquidity/client/lsps1.rs

@@ -488,7 +488,7 @@ impl LSPS1Liquidity {
 
  log_info!(self.logger, "Connected to LSP {}@{}. ", lsps1_node.node_id, lsps1_node.address);
 
- let refund_address = self.wallet.get_new_address()?;
+ let refund_address = self.runtime.block_on(self.wallet.get_new_address())?;
 
  let liquidity_source = Arc::clone(&self.liquidity_source);
  let response = self.runtime.block_on(async move {

src/payment/onchain.rs

@@ -15,6 +15,7 @@ use lightning::ln::channelmanager::PaymentId;
 use crate::config::Config;
 use crate::error::Error;
 use crate::logger::{log_info, LdkLogger, Logger};
+use crate::runtime::Runtime;
 use crate::types::{ChannelManager, Wallet};
 use crate::wallet::OnchainSendAmount;
 
@@ -47,23 +48,38 @@ pub struct OnchainPayment {
 	channel_manager: Arc<ChannelManager>,
 	config: Arc<Config>,
 	is_running: Arc<RwLock<bool>>,
+	runtime: Arc<Runtime>,
 	logger: Arc<Logger>,
 }
 
 impl OnchainPayment {
 	pub(crate) fn new(
  wallet: Arc<Wallet>, channel_manager: Arc<ChannelManager>, config: Arc<Config>,
- is_running: Arc<RwLock<bool>>, logger: Arc<Logger>,
+ is_running: Arc<RwLock<bool>>, runtime: Arc<Runtime>, logger: Arc<Logger>,
 	) -> Self {
- Self { wallet, channel_manager, config, is_running, logger }
+ Self { wallet, channel_manager, config, is_running, runtime, logger }
+	}
+
+	pub(crate) async fn send_to_address_inner(
+ &self, address: &bitcoin::Address, amount_sats: u64, fee_rate: Option<bitcoin::FeeRate>,
+	) -> Result<Txid, Error> {
+ if !*self.is_running.read().expect("lock") {
+ return Err(Error::NotRunning);
+ }
+
+ let cur_anchor_reserve_sats =
+ crate::total_anchor_channels_reserve_sats(&self.channel_manager, &self.config);
+ let send_amount =
+ OnchainSendAmount::ExactRetainingReserve { amount_sats, cur_anchor_reserve_sats };
+ self.wallet.send_to_address(address, send_amount, fee_rate).await
 	}
 }
 
 #[cfg_attr(feature = "uniffi", uniffi::export)]
 impl OnchainPayment {
 	/// Retrieve a new on-chain/funding address.
 	pub fn new_address(&self) -> Result<Address, Error> {
- let funding_address = self.wallet.get_new_address()?;
+ let funding_address = self.runtime.block_on(self.wallet.get_new_address())?;
  log_info!(self.logger, "Generated new funding address: {}", funding_address);
  Ok(funding_address)
 	}
@@ -80,16 +96,8 @@ impl OnchainPayment {
 	pub fn send_to_address(
  &self, address: &bitcoin::Address, amount_sats: u64, fee_rate: Option<FeeRate>,
 	) -> Result<Txid, Error> {
- if !*self.is_running.read().expect("lock") {
- return Err(Error::NotRunning);
- }
-
- let cur_anchor_reserve_sats =
- crate::total_anchor_channels_reserve_sats(&self.channel_manager, &self.config);
- let send_amount =
- OnchainSendAmount::ExactRetainingReserve { amount_sats, cur_anchor_reserve_sats };
  let fee_rate_opt = maybe_map_fee_rate_opt!(fee_rate);
- self.wallet.send_to_address(address, send_amount, fee_rate_opt)
+ self.runtime.block_on(self.send_to_address_inner(address, amount_sats, fee_rate_opt))
 	}
 
 	/// Send an on-chain payment to the given address, draining the available funds.
@@ -123,7 +131,7 @@ impl OnchainPayment {
  };
 
  let fee_rate_opt = maybe_map_fee_rate_opt!(fee_rate);
- self.wallet.send_to_address(address, send_amount, fee_rate_opt)
+ self.runtime.block_on(self.wallet.send_to_address(address, send_amount, fee_rate_opt))
 	}
 
 	/// Attempt to bump the fee of an unconfirmed transaction using Replace-by-Fee (RBF).
@@ -146,6 +154,10 @@ impl OnchainPayment {
  let cur_anchor_reserve_sats =
  crate::total_anchor_channels_reserve_sats(&self.channel_manager, &self.config);
  let fee_rate_opt = maybe_map_fee_rate_opt!(fee_rate);
- self.wallet.bump_fee_rbf(payment_id, fee_rate_opt, cur_anchor_reserve_sats)
+ self.runtime.block_on(self.wallet.bump_fee_rbf(
+ payment_id,
+ fee_rate_opt,
+ cur_anchor_reserve_sats,
+ ))
 	}
 }

src/payment/unified.rs

@@ -328,7 +328,10 @@ impl UnifiedPayment {
  Error::InvalidAmount
  })?;
 
- let txid = self.onchain_payment.send_to_address(&address, amt_sats, None)?;
+ let txid = self
+ .onchain_payment
+ .send_to_address_inner(&address, amt_sats, None)
+ .await?;
  return Ok(UnifiedPaymentResult::Onchain { txid });
  },
  }

src/chain/bitcoind.rs

@@ -461,11 +461,12 @@ impl BitcoindChainSource {
  evicted_txids.len(),
  elapsed_ms,
  );
- onchain_wallet.apply_mempool_txs(unconfirmed_txs, evicted_txids).unwrap_or_else(
- |e| {
+ onchain_wallet
+ .apply_mempool_txs(unconfirmed_txs, evicted_txids)
+ .await
+ .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/electrum.rs

@@ -182,7 +182,7 @@ impl ElectrumChainSource {
  update_res: Result<BdkUpdate, Error>, now: Instant,
 	) -> Result<(), Error> {
  match update_res {
- Ok(update) => match onchain_wallet.apply_update(update) {
+ Ok(update) => match onchain_wallet.apply_update(update).await {
  Ok(()) => {
  log_debug!(
  self.logger,

src/chain/esplora.rs

@@ -146,7 +146,7 @@ impl EsploraChainSource {
  let now = Instant::now();
  match $sync_future.await {
  Ok(res) => match res {
- Ok(update) => match onchain_wallet.apply_update(update) {
+ Ok(update) => match onchain_wallet.apply_update(update).await {
  Ok(()) => {
  log_debug!(
  self.logger,

src/event.rs

@@ -694,12 +694,16 @@ 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 funding_transaction = self
+ .wallet
+ .create_funding_transaction(
+ output_script,
+ channel_amount,
+ confirmation_target,
+ locktime,
+ )
+ .await;
+ match funding_transaction {
  Ok(final_tx) => {
  let needs_manual_broadcast = self
  .liquidity_source
@@ -1743,7 +1747,7 @@ where
  })
  .collect(),
  };
- if let Err(e) = self.wallet.cancel_tx(tx) {
+ if let Err(e) = self.wallet.cancel_tx(tx).await {
  log_error!(self.logger, "Failed reclaiming unused addresses: {}", e);
  return Err(ReplayEvent());
  }

src/lib.rs

@@ -1106,6 +1106,7 @@ impl Node {
  Arc::clone(&self.channel_manager),
  Arc::clone(&self.config),
  Arc::clone(&self.is_running),
+ Arc::clone(&self.runtime),
  Arc::clone(&self.logger),
  )
 	}
@@ -1118,6 +1119,7 @@ impl Node {
  Arc::clone(&self.channel_manager),
  Arc::clone(&self.config),
  Arc::clone(&self.is_running),
+ Arc::clone(&self.runtime),
  Arc::clone(&self.logger),
  ))
 	}

src/liquidity/client/lsps1.rs

@@ -488,7 +488,7 @@ impl LSPS1Liquidity {
 
  log_info!(self.logger, "Connected to LSP {}@{}. ", lsps1_node.node_id, lsps1_node.address);
 
- let refund_address = self.wallet.get_new_address()?;
+ let refund_address = self.runtime.block_on(self.wallet.get_new_address())?;
 
  let liquidity_source = Arc::clone(&self.liquidity_source);
  let response = self.runtime.block_on(async move {

src/payment/onchain.rs

@@ -15,6 +15,7 @@ use lightning::ln::channelmanager::PaymentId;
 use crate::config::Config;
 use crate::error::Error;
 use crate::logger::{log_info, LdkLogger, Logger};
+use crate::runtime::Runtime;
 use crate::types::{ChannelManager, Wallet};
 use crate::wallet::OnchainSendAmount;
 
@@ -47,23 +48,38 @@ pub struct OnchainPayment {
 	channel_manager: Arc<ChannelManager>,
 	config: Arc<Config>,
 	is_running: Arc<RwLock<bool>>,
+	runtime: Arc<Runtime>,
 	logger: Arc<Logger>,
 }
 
 impl OnchainPayment {
 	pub(crate) fn new(
  wallet: Arc<Wallet>, channel_manager: Arc<ChannelManager>, config: Arc<Config>,
- is_running: Arc<RwLock<bool>>, logger: Arc<Logger>,
+ is_running: Arc<RwLock<bool>>, runtime: Arc<Runtime>, logger: Arc<Logger>,
 	) -> Self {
- Self { wallet, channel_manager, config, is_running, logger }
+ Self { wallet, channel_manager, config, is_running, runtime, logger }
+	}
+
+	pub(crate) async fn send_to_address_inner(
+ &self, address: &bitcoin::Address, amount_sats: u64, fee_rate: Option<bitcoin::FeeRate>,
+	) -> Result<Txid, Error> {
+ if !*self.is_running.read().expect("lock") {
+ return Err(Error::NotRunning);
+ }
+
+ let cur_anchor_reserve_sats =
+ crate::total_anchor_channels_reserve_sats(&self.channel_manager, &self.config);
+ let send_amount =
+ OnchainSendAmount::ExactRetainingReserve { amount_sats, cur_anchor_reserve_sats };
+ self.wallet.send_to_address(address, send_amount, fee_rate).await
 	}
 }
 
 #[cfg_attr(feature = "uniffi", uniffi::export)]
 impl OnchainPayment {
 	/// Retrieve a new on-chain/funding address.
 	pub fn new_address(&self) -> Result<Address, Error> {
- let funding_address = self.wallet.get_new_address()?;
+ let funding_address = self.runtime.block_on(self.wallet.get_new_address())?;
  log_info!(self.logger, "Generated new funding address: {}", funding_address);
  Ok(funding_address)
 	}
@@ -80,16 +96,8 @@ impl OnchainPayment {
 	pub fn send_to_address(
  &self, address: &bitcoin::Address, amount_sats: u64, fee_rate: Option<FeeRate>,
 	) -> Result<Txid, Error> {
- if !*self.is_running.read().expect("lock") {
- return Err(Error::NotRunning);
- }
-
- let cur_anchor_reserve_sats =
- crate::total_anchor_channels_reserve_sats(&self.channel_manager, &self.config);
- let send_amount =
- OnchainSendAmount::ExactRetainingReserve { amount_sats, cur_anchor_reserve_sats };
  let fee_rate_opt = maybe_map_fee_rate_opt!(fee_rate);
- self.wallet.send_to_address(address, send_amount, fee_rate_opt)
+ self.runtime.block_on(self.send_to_address_inner(address, amount_sats, fee_rate_opt))
 	}
 
 	/// Send an on-chain payment to the given address, draining the available funds.
@@ -123,7 +131,7 @@ impl OnchainPayment {
  };
 
  let fee_rate_opt = maybe_map_fee_rate_opt!(fee_rate);
- self.wallet.send_to_address(address, send_amount, fee_rate_opt)
+ self.runtime.block_on(self.wallet.send_to_address(address, send_amount, fee_rate_opt))
 	}
 
 	/// Attempt to bump the fee of an unconfirmed transaction using Replace-by-Fee (RBF).
@@ -146,6 +154,10 @@ impl OnchainPayment {
  let cur_anchor_reserve_sats =
  crate::total_anchor_channels_reserve_sats(&self.channel_manager, &self.config);
  let fee_rate_opt = maybe_map_fee_rate_opt!(fee_rate);
- self.wallet.bump_fee_rbf(payment_id, fee_rate_opt, cur_anchor_reserve_sats)
+ self.runtime.block_on(self.wallet.bump_fee_rbf(
+ payment_id,
+ fee_rate_opt,
+ cur_anchor_reserve_sats,
+ ))
 	}
 }

src/payment/unified.rs

@@ -328,7 +328,10 @@ impl UnifiedPayment {
  Error::InvalidAmount
  })?;
 
- let txid = self.onchain_payment.send_to_address(&address, amt_sats, None)?;
+ let txid = self
+ .onchain_payment
+ .send_to_address_inner(&address, amt_sats, None)
+ .await?;
  return Ok(UnifiedPaymentResult::Onchain { txid });
  },
  }

src/chain/bitcoind.rs

@@ -461,11 +461,12 @@ impl BitcoindChainSource {
  evicted_txids.len(),
  elapsed_ms,
  );
- onchain_wallet.apply_mempool_txs(unconfirmed_txs, evicted_txids).unwrap_or_else(
- |e| {
+ onchain_wallet
+ .apply_mempool_txs(unconfirmed_txs, evicted_txids)
+ .await
+ .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/electrum.rs

@@ -182,7 +182,7 @@ impl ElectrumChainSource {
  update_res: Result<BdkUpdate, Error>, now: Instant,
 	) -> Result<(), Error> {
  match update_res {
- Ok(update) => match onchain_wallet.apply_update(update) {
+ Ok(update) => match onchain_wallet.apply_update(update).await {
  Ok(()) => {
  log_debug!(
  self.logger,

src/chain/esplora.rs

@@ -146,7 +146,7 @@ impl EsploraChainSource {
  let now = Instant::now();
  match $sync_future.await {
  Ok(res) => match res {
- Ok(update) => match onchain_wallet.apply_update(update) {
+ Ok(update) => match onchain_wallet.apply_update(update).await {
  Ok(()) => {
  log_debug!(
  self.logger,

src/event.rs

@@ -694,12 +694,16 @@ 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 funding_transaction = self
+ .wallet
+ .create_funding_transaction(
+ output_script,
+ channel_amount,
+ confirmation_target,
+ locktime,
+ )
+ .await;
+ match funding_transaction {
  Ok(final_tx) => {
  let needs_manual_broadcast = self
  .liquidity_source
@@ -1743,7 +1747,7 @@ where
  })
  .collect(),
  };
- if let Err(e) = self.wallet.cancel_tx(tx) {
+ if let Err(e) = self.wallet.cancel_tx(tx).await {
  log_error!(self.logger, "Failed reclaiming unused addresses: {}", e);
  return Err(ReplayEvent());
  }

src/lib.rs

@@ -1106,6 +1106,7 @@ impl Node {
  Arc::clone(&self.channel_manager),
  Arc::clone(&self.config),
  Arc::clone(&self.is_running),
+ Arc::clone(&self.runtime),
  Arc::clone(&self.logger),
  )
 	}
@@ -1118,6 +1119,7 @@ impl Node {
  Arc::clone(&self.channel_manager),
  Arc::clone(&self.config),
  Arc::clone(&self.is_running),
+ Arc::clone(&self.runtime),
  Arc::clone(&self.logger),
  ))
 	}

src/liquidity/client/lsps1.rs

@@ -488,7 +488,7 @@ impl LSPS1Liquidity {
 
  log_info!(self.logger, "Connected to LSP {}@{}. ", lsps1_node.node_id, lsps1_node.address);
 
- let refund_address = self.wallet.get_new_address()?;
+ let refund_address = self.runtime.block_on(self.wallet.get_new_address())?;
 
  let liquidity_source = Arc::clone(&self.liquidity_source);
  let response = self.runtime.block_on(async move {

src/payment/onchain.rs

@@ -15,6 +15,7 @@ use lightning::ln::channelmanager::PaymentId;
 use crate::config::Config;
 use crate::error::Error;
 use crate::logger::{log_info, LdkLogger, Logger};
+use crate::runtime::Runtime;
 use crate::types::{ChannelManager, Wallet};
 use crate::wallet::OnchainSendAmount;
 
@@ -47,23 +48,38 @@ pub struct OnchainPayment {
 	channel_manager: Arc<ChannelManager>,
 	config: Arc<Config>,
 	is_running: Arc<RwLock<bool>>,
+	runtime: Arc<Runtime>,
 	logger: Arc<Logger>,
 }
 
 impl OnchainPayment {
 	pub(crate) fn new(
  wallet: Arc<Wallet>, channel_manager: Arc<ChannelManager>, config: Arc<Config>,
- is_running: Arc<RwLock<bool>>, logger: Arc<Logger>,
+ is_running: Arc<RwLock<bool>>, runtime: Arc<Runtime>, logger: Arc<Logger>,
 	) -> Self {
- Self { wallet, channel_manager, config, is_running, logger }
+ Self { wallet, channel_manager, config, is_running, runtime, logger }
+	}
+
+	pub(crate) async fn send_to_address_inner(
+ &self, address: &bitcoin::Address, amount_sats: u64, fee_rate: Option<bitcoin::FeeRate>,
+	) -> Result<Txid, Error> {
+ if !*self.is_running.read().expect("lock") {
+ return Err(Error::NotRunning);
+ }
+
+ let cur_anchor_reserve_sats =
+ crate::total_anchor_channels_reserve_sats(&self.channel_manager, &self.config);
+ let send_amount =
+ OnchainSendAmount::ExactRetainingReserve { amount_sats, cur_anchor_reserve_sats };
+ self.wallet.send_to_address(address, send_amount, fee_rate).await
 	}
 }
 
 #[cfg_attr(feature = "uniffi", uniffi::export)]
 impl OnchainPayment {
 	/// Retrieve a new on-chain/funding address.
 	pub fn new_address(&self) -> Result<Address, Error> {
- let funding_address = self.wallet.get_new_address()?;
+ let funding_address = self.runtime.block_on(self.wallet.get_new_address())?;
  log_info!(self.logger, "Generated new funding address: {}", funding_address);
  Ok(funding_address)
 	}
@@ -80,16 +96,8 @@ impl OnchainPayment {
 	pub fn send_to_address(
  &self, address: &bitcoin::Address, amount_sats: u64, fee_rate: Option<FeeRate>,
 	) -> Result<Txid, Error> {
- if !*self.is_running.read().expect("lock") {
- return Err(Error::NotRunning);
- }
-
- let cur_anchor_reserve_sats =
- crate::total_anchor_channels_reserve_sats(&self.channel_manager, &self.config);
- let send_amount =
- OnchainSendAmount::ExactRetainingReserve { amount_sats, cur_anchor_reserve_sats };
  let fee_rate_opt = maybe_map_fee_rate_opt!(fee_rate);
- self.wallet.send_to_address(address, send_amount, fee_rate_opt)
+ self.runtime.block_on(self.send_to_address_inner(address, amount_sats, fee_rate_opt))
 	}
 
 	/// Send an on-chain payment to the given address, draining the available funds.
@@ -123,7 +131,7 @@ impl OnchainPayment {
  };
 
  let fee_rate_opt = maybe_map_fee_rate_opt!(fee_rate);
- self.wallet.send_to_address(address, send_amount, fee_rate_opt)
+ self.runtime.block_on(self.wallet.send_to_address(address, send_amount, fee_rate_opt))
 	}
 
 	/// Attempt to bump the fee of an unconfirmed transaction using Replace-by-Fee (RBF).
@@ -146,6 +154,10 @@ impl OnchainPayment {
  let cur_anchor_reserve_sats =
  crate::total_anchor_channels_reserve_sats(&self.channel_manager, &self.config);
  let fee_rate_opt = maybe_map_fee_rate_opt!(fee_rate);
- self.wallet.bump_fee_rbf(payment_id, fee_rate_opt, cur_anchor_reserve_sats)
+ self.runtime.block_on(self.wallet.bump_fee_rbf(
+ payment_id,
+ fee_rate_opt,
+ cur_anchor_reserve_sats,
+ ))
 	}
 }

src/payment/unified.rs

@@ -328,7 +328,10 @@ impl UnifiedPayment {
  Error::InvalidAmount
  })?;
 
- let txid = self.onchain_payment.send_to_address(&address, amt_sats, None)?;
+ let txid = self
+ .onchain_payment
+ .send_to_address_inner(&address, amt_sats, None)
+ .await?;
  return Ok(UnifiedPaymentResult::Onchain { txid });
  },
  }