Improve on (potentially) blocking wallet behaviors

Cargo.toml

@@ -64,7 +64,7 @@ bip39 = "2.0.0"
 
 rand = "0.8.5"
 chrono = { version = "0.4", default-features = false, features = ["clock"] }
-tokio = { version = "1", default-features = false, features = [ "rt-multi-thread", "time", "sync" ] }
+tokio = { version = "1.37", default-features = false, features = [ "rt-multi-thread", "time", "sync" ] }
 esplora-client = { version = "0.6", default-features = false }
 libc = "0.2"
 uniffi = { version = "0.26.0", features = ["build"], optional = true }

bindings/ldk_node.udl

@@ -165,11 +165,15 @@ enum NodeError {
 	"ChannelConfigUpdateFailed",
 	"PersistenceFailed",
 	"FeerateEstimationUpdateFailed",
+	"FeerateEstimationUpdateTimeout",
 	"WalletOperationFailed",
+	"WalletOperationTimeout",
 	"OnchainTxSigningFailed",
 	"MessageSigningFailed",
 	"TxSyncFailed",
+	"TxSyncTimeout",
 	"GossipUpdateFailed",
+	"GossipUpdateTimeout",
 	"LiquidityRequestFailed",
 	"InvalidAddress",
 	"InvalidSocketAddress",

src/builder.rs

@@ -1,6 +1,6 @@
 use crate::config::{
 	default_user_config, Config, BDK_CLIENT_CONCURRENCY, BDK_CLIENT_STOP_GAP,
-	DEFAULT_ESPLORA_SERVER_URL, WALLET_KEYS_SEED_LEN,
+DEFAULT_ESPLORA_CLIENT_TIMEOUT_SECS, DEFAULT_ESPLORA_SERVER_URL, WALLET_KEYS_SEED_LEN,
 };
 use crate::connection::ConnectionManager;
 use crate::event::EventQueue;
@@ -558,10 +558,15 @@ fn build_with_store_internal(
 
 	let (blockchain, tx_sync, tx_broadcaster, fee_estimator) = match chain_data_source_config {
  Some(ChainDataSourceConfig::Esplora(server_url)) => {
- let tx_sync = Arc::new(EsploraSyncClient::new(server_url.clone(), Arc::clone(&logger)));
- let blockchain =
- EsploraBlockchain::from_client(tx_sync.client().clone(), BDK_CLIENT_STOP_GAP)
- .with_concurrency(BDK_CLIENT_CONCURRENCY);
+ let mut client_builder = esplora_client::Builder::new(&server_url.clone());
+ client_builder = client_builder.timeout(DEFAULT_ESPLORA_CLIENT_TIMEOUT_SECS);
+ let esplora_client = client_builder.build_async().unwrap();
+ let tx_sync = Arc::new(EsploraSyncClient::from_client(
+ esplora_client.clone(),
+ Arc::clone(&logger),
+ ));
+ let blockchain = EsploraBlockchain::from_client(esplora_client, BDK_CLIENT_STOP_GAP)
+ .with_concurrency(BDK_CLIENT_CONCURRENCY);
  let tx_broadcaster = Arc::new(TransactionBroadcaster::new(
  tx_sync.client().clone(),
  Arc::clone(&logger),
@@ -959,6 +964,7 @@ fn build_with_store_internal(
 	};
 
 	let (stop_sender, _) = tokio::sync::watch::channel(());
+	let (event_handling_stopped_sender, _) = tokio::sync::watch::channel(());
 
 	let is_listening = Arc::new(AtomicBool::new(false));
 	let latest_wallet_sync_timestamp = Arc::new(RwLock::new(None));
@@ -971,6 +977,7 @@ fn build_with_store_internal(
 	Ok(Node {
  runtime,
  stop_sender,
+ event_handling_stopped_sender,
  config,
  wallet,
  tx_sync,

src/config.rs

@@ -28,6 +28,9 @@ pub(crate) const BDK_CLIENT_CONCURRENCY: u8 = 4;
 // The default Esplora server we're using.
 pub(crate) const DEFAULT_ESPLORA_SERVER_URL: &str = "https://blockstream.info/api";
 
+// The default Esplora client timeout we're using.
+pub(crate) const DEFAULT_ESPLORA_CLIENT_TIMEOUT_SECS: u64 = 10;
+
 // The timeout after which we abandon retrying failed payments.
 pub(crate) const LDK_PAYMENT_RETRY_TIMEOUT: Duration = Duration::from_secs(10);
 
@@ -46,6 +49,21 @@ pub(crate) const NODE_ANN_BCAST_INTERVAL: Duration = Duration::from_secs(60 * 60
 // The lower limit which we apply to any configured wallet sync intervals.
 pub(crate) const WALLET_SYNC_INTERVAL_MINIMUM_SECS: u64 = 10;
 
+// The timeout after which we abort a wallet syncing operation.
+pub(crate) const BDK_WALLET_SYNC_TIMEOUT_SECS: u64 = 90;
+
+// The timeout after which we abort a wallet syncing operation.
+pub(crate) const LDK_WALLET_SYNC_TIMEOUT_SECS: u64 = 30;
+
+// The timeout after which we abort a fee rate cache update operation.
+pub(crate) const FEE_RATE_CACHE_UPDATE_TIMEOUT_SECS: u64 = 5;
+
+// The timeout after which we abort a transaction broadcast operation.
+pub(crate) const TX_BROADCAST_TIMEOUT_SECS: u64 = 5;
+
+// The timeout after which we abort a RGS sync operation.
+pub(crate) const RGS_SYNC_TIMEOUT_SECS: u64 = 5;
+
 // The length in bytes of our wallets' keys seed.
 pub(crate) const WALLET_KEYS_SEED_LEN: usize = 64;
 

src/error.rs

@@ -33,16 +33,24 @@ pub enum Error {
 	PersistenceFailed,
 	/// A fee rate estimation update failed.
 	FeerateEstimationUpdateFailed,
+	/// A fee rate estimation update timed out.
+	FeerateEstimationUpdateTimeout,
 	/// A wallet operation failed.
 	WalletOperationFailed,
+	/// A wallet operation timed out.
+	WalletOperationTimeout,
 	/// A signing operation for transaction failed.
 	OnchainTxSigningFailed,
 	/// A signing operation for message failed.
 	MessageSigningFailed,
 	/// A transaction sync operation failed.
 	TxSyncFailed,
+	/// A transaction sync operation timed out.
+	TxSyncTimeout,
 	/// A gossip updating operation failed.
 	GossipUpdateFailed,
+	/// A gossip updating operation timed out.
+	GossipUpdateTimeout,
 	/// A liquidity request operation failed.
 	LiquidityRequestFailed,
 	/// The given address is invalid.
@@ -111,11 +119,17 @@ impl fmt::Display for Error {
  Self::FeerateEstimationUpdateFailed => {
  write!(f, "Failed to update fee rate estimates.")
  },
+ Self::FeerateEstimationUpdateTimeout => {
+ write!(f, "Updating fee rate estimates timed out.")
+ },
  Self::WalletOperationFailed => write!(f, "Failed to conduct wallet operation."),
+ Self::WalletOperationTimeout => write!(f, "A wallet operation timed out."),
  Self::OnchainTxSigningFailed => write!(f, "Failed to sign given transaction."),
  Self::MessageSigningFailed => write!(f, "Failed to sign given message."),
  Self::TxSyncFailed => write!(f, "Failed to sync transactions."),
+ Self::TxSyncTimeout => write!(f, "Syncing transactions timed out."),
  Self::GossipUpdateFailed => write!(f, "Failed to update gossip data."),
+ Self::GossipUpdateTimeout => write!(f, "Updating gossip data timed out."),
  Self::LiquidityRequestFailed => write!(f, "Failed to request inbound liquidity."),
  Self::InvalidAddress => write!(f, "The given address is invalid."),
  Self::InvalidSocketAddress => write!(f, "The given network address is invalid."),

src/fee_estimator.rs

@@ -1,3 +1,4 @@
+use crate::config::FEE_RATE_CACHE_UPDATE_TIMEOUT_SECS;
 use crate::logger::{log_error, log_trace, Logger};
 use crate::{Config, Error};
 
@@ -14,6 +15,7 @@ use bitcoin::Network;
 use std::collections::HashMap;
 use std::ops::Deref;
 use std::sync::{Arc, RwLock};
+use std::time::Duration;
 
 pub(crate) struct OnchainFeeEstimator<L: Deref>
 where
@@ -55,7 +57,21 @@ where
  ConfirmationTarget::OutputSpendingFee => 12,
  };
 
- let estimates = self.esplora_client.get_fee_estimates().await.map_err(|e| {
+ let estimates = tokio::time::timeout(
+ Duration::from_secs(FEE_RATE_CACHE_UPDATE_TIMEOUT_SECS),
+ self.esplora_client.get_fee_estimates(),
+ )
+ .await
+ .map_err(|e| {
+ log_error!(
+ self.logger,
+ "Updating fee rate estimates for {:?} timed out: {}",
+ target,
+ e
+ );
+ Error::FeerateEstimationUpdateTimeout
+ })?
+ .map_err(|e| {
  log_error!(
  self.logger,
  "Failed to retrieve fee rate estimates for {:?}: {}",

src/gossip.rs

@@ -1,3 +1,4 @@
+use crate::config::RGS_SYNC_TIMEOUT_SECS;
 use crate::logger::{log_trace, FilesystemLogger, Logger};
 use crate::types::{GossipSync, Graph, P2PGossipSync, RapidGossipSync};
 use crate::Error;
@@ -6,6 +7,7 @@ use lightning::routing::utxo::UtxoLookup;
 
 use std::sync::atomic::{AtomicU32, Ordering};
 use std::sync::Arc;
+use std::time::Duration;
 
 pub(crate) enum GossipSource {
 	P2PNetwork {
@@ -55,7 +57,17 @@ impl GossipSource {
  Self::RapidGossipSync { gossip_sync, server_url, latest_sync_timestamp, logger } => {
  let query_timestamp = latest_sync_timestamp.load(Ordering::Acquire);
  let query_url = format!("{}/{}", server_url, query_timestamp);
- let response = reqwest::get(query_url).await.map_err(|e| {
+
+ let response = tokio::time::timeout(
+ Duration::from_secs(RGS_SYNC_TIMEOUT_SECS),
+ reqwest::get(query_url),
+ )
+ .await
+ .map_err(|e| {
+ log_trace!(logger, "Retrieving RGS gossip update timed out: {}", e);
+ Error::GossipUpdateTimeout
+ })?
+ .map_err(|e| {
  log_trace!(logger, "Failed to retrieve RGS gossip update: {}", e);
  Error::GossipUpdateFailed
  })?;

Cargo.toml

@@ -64,7 +64,7 @@ bip39 = "2.0.0"
 
 rand = "0.8.5"
 chrono = { version = "0.4", default-features = false, features = ["clock"] }
-tokio = { version = "1", default-features = false, features = [ "rt-multi-thread", "time", "sync" ] }
+tokio = { version = "1.37", default-features = false, features = [ "rt-multi-thread", "time", "sync" ] }
 esplora-client = { version = "0.6", default-features = false }
 libc = "0.2"
 uniffi = { version = "0.26.0", features = ["build"], optional = true }

bindings/ldk_node.udl

@@ -165,11 +165,15 @@ enum NodeError {
 	"ChannelConfigUpdateFailed",
 	"PersistenceFailed",
 	"FeerateEstimationUpdateFailed",
+	"FeerateEstimationUpdateTimeout",
 	"WalletOperationFailed",
+	"WalletOperationTimeout",
 	"OnchainTxSigningFailed",
 	"MessageSigningFailed",
 	"TxSyncFailed",
+	"TxSyncTimeout",
 	"GossipUpdateFailed",
+	"GossipUpdateTimeout",
 	"LiquidityRequestFailed",
 	"InvalidAddress",
 	"InvalidSocketAddress",

src/builder.rs

@@ -1,6 +1,6 @@
 use crate::config::{
 	default_user_config, Config, BDK_CLIENT_CONCURRENCY, BDK_CLIENT_STOP_GAP,
-	DEFAULT_ESPLORA_SERVER_URL, WALLET_KEYS_SEED_LEN,
+DEFAULT_ESPLORA_CLIENT_TIMEOUT_SECS, DEFAULT_ESPLORA_SERVER_URL, WALLET_KEYS_SEED_LEN,
 };
 use crate::connection::ConnectionManager;
 use crate::event::EventQueue;
@@ -558,10 +558,15 @@ fn build_with_store_internal(
 
 	let (blockchain, tx_sync, tx_broadcaster, fee_estimator) = match chain_data_source_config {
  Some(ChainDataSourceConfig::Esplora(server_url)) => {
- let tx_sync = Arc::new(EsploraSyncClient::new(server_url.clone(), Arc::clone(&logger)));
- let blockchain =
- EsploraBlockchain::from_client(tx_sync.client().clone(), BDK_CLIENT_STOP_GAP)
- .with_concurrency(BDK_CLIENT_CONCURRENCY);
+ let mut client_builder = esplora_client::Builder::new(&server_url.clone());
+ client_builder = client_builder.timeout(DEFAULT_ESPLORA_CLIENT_TIMEOUT_SECS);
+ let esplora_client = client_builder.build_async().unwrap();
+ let tx_sync = Arc::new(EsploraSyncClient::from_client(
+ esplora_client.clone(),
+ Arc::clone(&logger),
+ ));
+ let blockchain = EsploraBlockchain::from_client(esplora_client, BDK_CLIENT_STOP_GAP)
+ .with_concurrency(BDK_CLIENT_CONCURRENCY);
  let tx_broadcaster = Arc::new(TransactionBroadcaster::new(
  tx_sync.client().clone(),
  Arc::clone(&logger),
@@ -959,6 +964,7 @@ fn build_with_store_internal(
 	};
 
 	let (stop_sender, _) = tokio::sync::watch::channel(());
+	let (event_handling_stopped_sender, _) = tokio::sync::watch::channel(());
 
 	let is_listening = Arc::new(AtomicBool::new(false));
 	let latest_wallet_sync_timestamp = Arc::new(RwLock::new(None));
@@ -971,6 +977,7 @@ fn build_with_store_internal(
 	Ok(Node {
  runtime,
  stop_sender,
+ event_handling_stopped_sender,
  config,
  wallet,
  tx_sync,

src/config.rs

@@ -28,6 +28,9 @@ pub(crate) const BDK_CLIENT_CONCURRENCY: u8 = 4;
 // The default Esplora server we're using.
 pub(crate) const DEFAULT_ESPLORA_SERVER_URL: &str = "https://blockstream.info/api";
 
+// The default Esplora client timeout we're using.
+pub(crate) const DEFAULT_ESPLORA_CLIENT_TIMEOUT_SECS: u64 = 10;
+
 // The timeout after which we abandon retrying failed payments.
 pub(crate) const LDK_PAYMENT_RETRY_TIMEOUT: Duration = Duration::from_secs(10);
 
@@ -46,6 +49,21 @@ pub(crate) const NODE_ANN_BCAST_INTERVAL: Duration = Duration::from_secs(60 * 60
 // The lower limit which we apply to any configured wallet sync intervals.
 pub(crate) const WALLET_SYNC_INTERVAL_MINIMUM_SECS: u64 = 10;
 
+// The timeout after which we abort a wallet syncing operation.
+pub(crate) const BDK_WALLET_SYNC_TIMEOUT_SECS: u64 = 90;
+
+// The timeout after which we abort a wallet syncing operation.
+pub(crate) const LDK_WALLET_SYNC_TIMEOUT_SECS: u64 = 30;
+
+// The timeout after which we abort a fee rate cache update operation.
+pub(crate) const FEE_RATE_CACHE_UPDATE_TIMEOUT_SECS: u64 = 5;
+
+// The timeout after which we abort a transaction broadcast operation.
+pub(crate) const TX_BROADCAST_TIMEOUT_SECS: u64 = 5;
+
+// The timeout after which we abort a RGS sync operation.
+pub(crate) const RGS_SYNC_TIMEOUT_SECS: u64 = 5;
+
 // The length in bytes of our wallets' keys seed.
 pub(crate) const WALLET_KEYS_SEED_LEN: usize = 64;
 

src/error.rs

@@ -33,16 +33,24 @@ pub enum Error {
 	PersistenceFailed,
 	/// A fee rate estimation update failed.
 	FeerateEstimationUpdateFailed,
+	/// A fee rate estimation update timed out.
+	FeerateEstimationUpdateTimeout,
 	/// A wallet operation failed.
 	WalletOperationFailed,
+	/// A wallet operation timed out.
+	WalletOperationTimeout,
 	/// A signing operation for transaction failed.
 	OnchainTxSigningFailed,
 	/// A signing operation for message failed.
 	MessageSigningFailed,
 	/// A transaction sync operation failed.
 	TxSyncFailed,
+	/// A transaction sync operation timed out.
+	TxSyncTimeout,
 	/// A gossip updating operation failed.
 	GossipUpdateFailed,
+	/// A gossip updating operation timed out.
+	GossipUpdateTimeout,
 	/// A liquidity request operation failed.
 	LiquidityRequestFailed,
 	/// The given address is invalid.
@@ -111,11 +119,17 @@ impl fmt::Display for Error {
  Self::FeerateEstimationUpdateFailed => {
  write!(f, "Failed to update fee rate estimates.")
  },
+ Self::FeerateEstimationUpdateTimeout => {
+ write!(f, "Updating fee rate estimates timed out.")
+ },
  Self::WalletOperationFailed => write!(f, "Failed to conduct wallet operation."),
+ Self::WalletOperationTimeout => write!(f, "A wallet operation timed out."),
  Self::OnchainTxSigningFailed => write!(f, "Failed to sign given transaction."),
  Self::MessageSigningFailed => write!(f, "Failed to sign given message."),
  Self::TxSyncFailed => write!(f, "Failed to sync transactions."),
+ Self::TxSyncTimeout => write!(f, "Syncing transactions timed out."),
  Self::GossipUpdateFailed => write!(f, "Failed to update gossip data."),
+ Self::GossipUpdateTimeout => write!(f, "Updating gossip data timed out."),
  Self::LiquidityRequestFailed => write!(f, "Failed to request inbound liquidity."),
  Self::InvalidAddress => write!(f, "The given address is invalid."),
  Self::InvalidSocketAddress => write!(f, "The given network address is invalid."),

src/fee_estimator.rs

@@ -1,3 +1,4 @@
+use crate::config::FEE_RATE_CACHE_UPDATE_TIMEOUT_SECS;
 use crate::logger::{log_error, log_trace, Logger};
 use crate::{Config, Error};
 
@@ -14,6 +15,7 @@ use bitcoin::Network;
 use std::collections::HashMap;
 use std::ops::Deref;
 use std::sync::{Arc, RwLock};
+use std::time::Duration;
 
 pub(crate) struct OnchainFeeEstimator<L: Deref>
 where
@@ -55,7 +57,21 @@ where
  ConfirmationTarget::OutputSpendingFee => 12,
  };
 
- let estimates = self.esplora_client.get_fee_estimates().await.map_err(|e| {
+ let estimates = tokio::time::timeout(
+ Duration::from_secs(FEE_RATE_CACHE_UPDATE_TIMEOUT_SECS),
+ self.esplora_client.get_fee_estimates(),
+ )
+ .await
+ .map_err(|e| {
+ log_error!(
+ self.logger,
+ "Updating fee rate estimates for {:?} timed out: {}",
+ target,
+ e
+ );
+ Error::FeerateEstimationUpdateTimeout
+ })?
+ .map_err(|e| {
  log_error!(
  self.logger,
  "Failed to retrieve fee rate estimates for {:?}: {}",

src/gossip.rs

@@ -1,3 +1,4 @@
+use crate::config::RGS_SYNC_TIMEOUT_SECS;
 use crate::logger::{log_trace, FilesystemLogger, Logger};
 use crate::types::{GossipSync, Graph, P2PGossipSync, RapidGossipSync};
 use crate::Error;
@@ -6,6 +7,7 @@ use lightning::routing::utxo::UtxoLookup;
 
 use std::sync::atomic::{AtomicU32, Ordering};
 use std::sync::Arc;
+use std::time::Duration;
 
 pub(crate) enum GossipSource {
 	P2PNetwork {
@@ -55,7 +57,17 @@ impl GossipSource {
  Self::RapidGossipSync { gossip_sync, server_url, latest_sync_timestamp, logger } => {
  let query_timestamp = latest_sync_timestamp.load(Ordering::Acquire);
  let query_url = format!("{}/{}", server_url, query_timestamp);
- let response = reqwest::get(query_url).await.map_err(|e| {
+
+ let response = tokio::time::timeout(
+ Duration::from_secs(RGS_SYNC_TIMEOUT_SECS),
+ reqwest::get(query_url),
+ )
+ .await
+ .map_err(|e| {
+ log_trace!(logger, "Retrieving RGS gossip update timed out: {}", e);
+ Error::GossipUpdateTimeout
+ })?
+ .map_err(|e| {
  log_trace!(logger, "Failed to retrieve RGS gossip update: {}", e);
  Error::GossipUpdateFailed
  })?;

Cargo.toml

@@ -64,7 +64,7 @@ bip39 = "2.0.0"
 
 rand = "0.8.5"
 chrono = { version = "0.4", default-features = false, features = ["clock"] }
-tokio = { version = "1", default-features = false, features = [ "rt-multi-thread", "time", "sync" ] }
+tokio = { version = "1.37", default-features = false, features = [ "rt-multi-thread", "time", "sync" ] }
 esplora-client = { version = "0.6", default-features = false }
 libc = "0.2"
 uniffi = { version = "0.26.0", features = ["build"], optional = true }

bindings/ldk_node.udl

@@ -165,11 +165,15 @@ enum NodeError {
 	"ChannelConfigUpdateFailed",
 	"PersistenceFailed",
 	"FeerateEstimationUpdateFailed",
+	"FeerateEstimationUpdateTimeout",
 	"WalletOperationFailed",
+	"WalletOperationTimeout",
 	"OnchainTxSigningFailed",
 	"MessageSigningFailed",
 	"TxSyncFailed",
+	"TxSyncTimeout",
 	"GossipUpdateFailed",
+	"GossipUpdateTimeout",
 	"LiquidityRequestFailed",
 	"InvalidAddress",
 	"InvalidSocketAddress",

src/builder.rs

@@ -1,6 +1,6 @@
 use crate::config::{
 	default_user_config, Config, BDK_CLIENT_CONCURRENCY, BDK_CLIENT_STOP_GAP,
-	DEFAULT_ESPLORA_SERVER_URL, WALLET_KEYS_SEED_LEN,
+DEFAULT_ESPLORA_CLIENT_TIMEOUT_SECS, DEFAULT_ESPLORA_SERVER_URL, WALLET_KEYS_SEED_LEN,
 };
 use crate::connection::ConnectionManager;
 use crate::event::EventQueue;
@@ -558,10 +558,15 @@ fn build_with_store_internal(
 
 	let (blockchain, tx_sync, tx_broadcaster, fee_estimator) = match chain_data_source_config {
  Some(ChainDataSourceConfig::Esplora(server_url)) => {
- let tx_sync = Arc::new(EsploraSyncClient::new(server_url.clone(), Arc::clone(&logger)));
- let blockchain =
- EsploraBlockchain::from_client(tx_sync.client().clone(), BDK_CLIENT_STOP_GAP)
- .with_concurrency(BDK_CLIENT_CONCURRENCY);
+ let mut client_builder = esplora_client::Builder::new(&server_url.clone());
+ client_builder = client_builder.timeout(DEFAULT_ESPLORA_CLIENT_TIMEOUT_SECS);
+ let esplora_client = client_builder.build_async().unwrap();
+ let tx_sync = Arc::new(EsploraSyncClient::from_client(
+ esplora_client.clone(),
+ Arc::clone(&logger),
+ ));
+ let blockchain = EsploraBlockchain::from_client(esplora_client, BDK_CLIENT_STOP_GAP)
+ .with_concurrency(BDK_CLIENT_CONCURRENCY);
  let tx_broadcaster = Arc::new(TransactionBroadcaster::new(
  tx_sync.client().clone(),
  Arc::clone(&logger),
@@ -959,6 +964,7 @@ fn build_with_store_internal(
 	};
 
 	let (stop_sender, _) = tokio::sync::watch::channel(());
+	let (event_handling_stopped_sender, _) = tokio::sync::watch::channel(());
 
 	let is_listening = Arc::new(AtomicBool::new(false));
 	let latest_wallet_sync_timestamp = Arc::new(RwLock::new(None));
@@ -971,6 +977,7 @@ fn build_with_store_internal(
 	Ok(Node {
  runtime,
  stop_sender,
+ event_handling_stopped_sender,
  config,
  wallet,
  tx_sync,

src/config.rs

@@ -28,6 +28,9 @@ pub(crate) const BDK_CLIENT_CONCURRENCY: u8 = 4;
 // The default Esplora server we're using.
 pub(crate) const DEFAULT_ESPLORA_SERVER_URL: &str = "https://blockstream.info/api";
 
+// The default Esplora client timeout we're using.
+pub(crate) const DEFAULT_ESPLORA_CLIENT_TIMEOUT_SECS: u64 = 10;
+
 // The timeout after which we abandon retrying failed payments.
 pub(crate) const LDK_PAYMENT_RETRY_TIMEOUT: Duration = Duration::from_secs(10);
 
@@ -46,6 +49,21 @@ pub(crate) const NODE_ANN_BCAST_INTERVAL: Duration = Duration::from_secs(60 * 60
 // The lower limit which we apply to any configured wallet sync intervals.
 pub(crate) const WALLET_SYNC_INTERVAL_MINIMUM_SECS: u64 = 10;
 
+// The timeout after which we abort a wallet syncing operation.
+pub(crate) const BDK_WALLET_SYNC_TIMEOUT_SECS: u64 = 90;
+
+// The timeout after which we abort a wallet syncing operation.
+pub(crate) const LDK_WALLET_SYNC_TIMEOUT_SECS: u64 = 30;
+
+// The timeout after which we abort a fee rate cache update operation.
+pub(crate) const FEE_RATE_CACHE_UPDATE_TIMEOUT_SECS: u64 = 5;
+
+// The timeout after which we abort a transaction broadcast operation.
+pub(crate) const TX_BROADCAST_TIMEOUT_SECS: u64 = 5;
+
+// The timeout after which we abort a RGS sync operation.
+pub(crate) const RGS_SYNC_TIMEOUT_SECS: u64 = 5;
+
 // The length in bytes of our wallets' keys seed.
 pub(crate) const WALLET_KEYS_SEED_LEN: usize = 64;
 

src/error.rs

@@ -33,16 +33,24 @@ pub enum Error {
 	PersistenceFailed,
 	/// A fee rate estimation update failed.
 	FeerateEstimationUpdateFailed,
+	/// A fee rate estimation update timed out.
+	FeerateEstimationUpdateTimeout,
 	/// A wallet operation failed.
 	WalletOperationFailed,
+	/// A wallet operation timed out.
+	WalletOperationTimeout,
 	/// A signing operation for transaction failed.
 	OnchainTxSigningFailed,
 	/// A signing operation for message failed.
 	MessageSigningFailed,
 	/// A transaction sync operation failed.
 	TxSyncFailed,
+	/// A transaction sync operation timed out.
+	TxSyncTimeout,
 	/// A gossip updating operation failed.
 	GossipUpdateFailed,
+	/// A gossip updating operation timed out.
+	GossipUpdateTimeout,
 	/// A liquidity request operation failed.
 	LiquidityRequestFailed,
 	/// The given address is invalid.
@@ -111,11 +119,17 @@ impl fmt::Display for Error {
  Self::FeerateEstimationUpdateFailed => {
  write!(f, "Failed to update fee rate estimates.")
  },
+ Self::FeerateEstimationUpdateTimeout => {
+ write!(f, "Updating fee rate estimates timed out.")
+ },
  Self::WalletOperationFailed => write!(f, "Failed to conduct wallet operation."),
+ Self::WalletOperationTimeout => write!(f, "A wallet operation timed out."),
  Self::OnchainTxSigningFailed => write!(f, "Failed to sign given transaction."),
  Self::MessageSigningFailed => write!(f, "Failed to sign given message."),
  Self::TxSyncFailed => write!(f, "Failed to sync transactions."),
+ Self::TxSyncTimeout => write!(f, "Syncing transactions timed out."),
  Self::GossipUpdateFailed => write!(f, "Failed to update gossip data."),
+ Self::GossipUpdateTimeout => write!(f, "Updating gossip data timed out."),
  Self::LiquidityRequestFailed => write!(f, "Failed to request inbound liquidity."),
  Self::InvalidAddress => write!(f, "The given address is invalid."),
  Self::InvalidSocketAddress => write!(f, "The given network address is invalid."),

src/fee_estimator.rs

@@ -1,3 +1,4 @@
+use crate::config::FEE_RATE_CACHE_UPDATE_TIMEOUT_SECS;
 use crate::logger::{log_error, log_trace, Logger};
 use crate::{Config, Error};
 
@@ -14,6 +15,7 @@ use bitcoin::Network;
 use std::collections::HashMap;
 use std::ops::Deref;
 use std::sync::{Arc, RwLock};
+use std::time::Duration;
 
 pub(crate) struct OnchainFeeEstimator<L: Deref>
 where
@@ -55,7 +57,21 @@ where
  ConfirmationTarget::OutputSpendingFee => 12,
  };
 
- let estimates = self.esplora_client.get_fee_estimates().await.map_err(|e| {
+ let estimates = tokio::time::timeout(
+ Duration::from_secs(FEE_RATE_CACHE_UPDATE_TIMEOUT_SECS),
+ self.esplora_client.get_fee_estimates(),
+ )
+ .await
+ .map_err(|e| {
+ log_error!(
+ self.logger,
+ "Updating fee rate estimates for {:?} timed out: {}",
+ target,
+ e
+ );
+ Error::FeerateEstimationUpdateTimeout
+ })?
+ .map_err(|e| {
  log_error!(
  self.logger,
  "Failed to retrieve fee rate estimates for {:?}: {}",

src/gossip.rs

@@ -1,3 +1,4 @@
+use crate::config::RGS_SYNC_TIMEOUT_SECS;
 use crate::logger::{log_trace, FilesystemLogger, Logger};
 use crate::types::{GossipSync, Graph, P2PGossipSync, RapidGossipSync};
 use crate::Error;
@@ -6,6 +7,7 @@ use lightning::routing::utxo::UtxoLookup;
 
 use std::sync::atomic::{AtomicU32, Ordering};
 use std::sync::Arc;
+use std::time::Duration;
 
 pub(crate) enum GossipSource {
 	P2PNetwork {
@@ -55,7 +57,17 @@ impl GossipSource {
  Self::RapidGossipSync { gossip_sync, server_url, latest_sync_timestamp, logger } => {
  let query_timestamp = latest_sync_timestamp.load(Ordering::Acquire);
  let query_url = format!("{}/{}", server_url, query_timestamp);
- let response = reqwest::get(query_url).await.map_err(|e| {
+
+ let response = tokio::time::timeout(
+ Duration::from_secs(RGS_SYNC_TIMEOUT_SECS),
+ reqwest::get(query_url),
+ )
+ .await
+ .map_err(|e| {
+ log_trace!(logger, "Retrieving RGS gossip update timed out: {}", e);
+ Error::GossipUpdateTimeout
+ })?
+ .map_err(|e| {
  log_trace!(logger, "Failed to retrieve RGS gossip update: {}", e);
  Error::GossipUpdateFailed
  })?;

Cargo.toml

@@ -64,7 +64,7 @@ bip39 = "2.0.0"
 
 rand = "0.8.5"
 chrono = { version = "0.4", default-features = false, features = ["clock"] }
-tokio = { version = "1", default-features = false, features = [ "rt-multi-thread", "time", "sync" ] }
+tokio = { version = "1.37", default-features = false, features = [ "rt-multi-thread", "time", "sync" ] }
 esplora-client = { version = "0.6", default-features = false }
 libc = "0.2"
 uniffi = { version = "0.26.0", features = ["build"], optional = true }

bindings/ldk_node.udl

@@ -165,11 +165,15 @@ enum NodeError {
 	"ChannelConfigUpdateFailed",
 	"PersistenceFailed",
 	"FeerateEstimationUpdateFailed",
+	"FeerateEstimationUpdateTimeout",
 	"WalletOperationFailed",
+	"WalletOperationTimeout",
 	"OnchainTxSigningFailed",
 	"MessageSigningFailed",
 	"TxSyncFailed",
+	"TxSyncTimeout",
 	"GossipUpdateFailed",
+	"GossipUpdateTimeout",
 	"LiquidityRequestFailed",
 	"InvalidAddress",
 	"InvalidSocketAddress",

src/builder.rs

@@ -1,6 +1,6 @@
 use crate::config::{
 	default_user_config, Config, BDK_CLIENT_CONCURRENCY, BDK_CLIENT_STOP_GAP,
-	DEFAULT_ESPLORA_SERVER_URL, WALLET_KEYS_SEED_LEN,
+DEFAULT_ESPLORA_CLIENT_TIMEOUT_SECS, DEFAULT_ESPLORA_SERVER_URL, WALLET_KEYS_SEED_LEN,
 };
 use crate::connection::ConnectionManager;
 use crate::event::EventQueue;
@@ -558,10 +558,15 @@ fn build_with_store_internal(
 
 	let (blockchain, tx_sync, tx_broadcaster, fee_estimator) = match chain_data_source_config {
  Some(ChainDataSourceConfig::Esplora(server_url)) => {
- let tx_sync = Arc::new(EsploraSyncClient::new(server_url.clone(), Arc::clone(&logger)));
- let blockchain =
- EsploraBlockchain::from_client(tx_sync.client().clone(), BDK_CLIENT_STOP_GAP)
- .with_concurrency(BDK_CLIENT_CONCURRENCY);
+ let mut client_builder = esplora_client::Builder::new(&server_url.clone());
+ client_builder = client_builder.timeout(DEFAULT_ESPLORA_CLIENT_TIMEOUT_SECS);
+ let esplora_client = client_builder.build_async().unwrap();
+ let tx_sync = Arc::new(EsploraSyncClient::from_client(
+ esplora_client.clone(),
+ Arc::clone(&logger),
+ ));
+ let blockchain = EsploraBlockchain::from_client(esplora_client, BDK_CLIENT_STOP_GAP)
+ .with_concurrency(BDK_CLIENT_CONCURRENCY);
  let tx_broadcaster = Arc::new(TransactionBroadcaster::new(
  tx_sync.client().clone(),
  Arc::clone(&logger),
@@ -959,6 +964,7 @@ fn build_with_store_internal(
 	};
 
 	let (stop_sender, _) = tokio::sync::watch::channel(());
+	let (event_handling_stopped_sender, _) = tokio::sync::watch::channel(());
 
 	let is_listening = Arc::new(AtomicBool::new(false));
 	let latest_wallet_sync_timestamp = Arc::new(RwLock::new(None));
@@ -971,6 +977,7 @@ fn build_with_store_internal(
 	Ok(Node {
  runtime,
  stop_sender,
+ event_handling_stopped_sender,
  config,
  wallet,
  tx_sync,

src/config.rs

@@ -28,6 +28,9 @@ pub(crate) const BDK_CLIENT_CONCURRENCY: u8 = 4;
 // The default Esplora server we're using.
 pub(crate) const DEFAULT_ESPLORA_SERVER_URL: &str = "https://blockstream.info/api";
 
+// The default Esplora client timeout we're using.
+pub(crate) const DEFAULT_ESPLORA_CLIENT_TIMEOUT_SECS: u64 = 10;
+
 // The timeout after which we abandon retrying failed payments.
 pub(crate) const LDK_PAYMENT_RETRY_TIMEOUT: Duration = Duration::from_secs(10);
 
@@ -46,6 +49,21 @@ pub(crate) const NODE_ANN_BCAST_INTERVAL: Duration = Duration::from_secs(60 * 60
 // The lower limit which we apply to any configured wallet sync intervals.
 pub(crate) const WALLET_SYNC_INTERVAL_MINIMUM_SECS: u64 = 10;
 
+// The timeout after which we abort a wallet syncing operation.
+pub(crate) const BDK_WALLET_SYNC_TIMEOUT_SECS: u64 = 90;
+
+// The timeout after which we abort a wallet syncing operation.
+pub(crate) const LDK_WALLET_SYNC_TIMEOUT_SECS: u64 = 30;
+
+// The timeout after which we abort a fee rate cache update operation.
+pub(crate) const FEE_RATE_CACHE_UPDATE_TIMEOUT_SECS: u64 = 5;
+
+// The timeout after which we abort a transaction broadcast operation.
+pub(crate) const TX_BROADCAST_TIMEOUT_SECS: u64 = 5;
+
+// The timeout after which we abort a RGS sync operation.
+pub(crate) const RGS_SYNC_TIMEOUT_SECS: u64 = 5;
+
 // The length in bytes of our wallets' keys seed.
 pub(crate) const WALLET_KEYS_SEED_LEN: usize = 64;
 

src/error.rs

@@ -33,16 +33,24 @@ pub enum Error {
 	PersistenceFailed,
 	/// A fee rate estimation update failed.
 	FeerateEstimationUpdateFailed,
+	/// A fee rate estimation update timed out.
+	FeerateEstimationUpdateTimeout,
 	/// A wallet operation failed.
 	WalletOperationFailed,
+	/// A wallet operation timed out.
+	WalletOperationTimeout,
 	/// A signing operation for transaction failed.
 	OnchainTxSigningFailed,
 	/// A signing operation for message failed.
 	MessageSigningFailed,
 	/// A transaction sync operation failed.
 	TxSyncFailed,
+	/// A transaction sync operation timed out.
+	TxSyncTimeout,
 	/// A gossip updating operation failed.
 	GossipUpdateFailed,
+	/// A gossip updating operation timed out.
+	GossipUpdateTimeout,
 	/// A liquidity request operation failed.
 	LiquidityRequestFailed,
 	/// The given address is invalid.
@@ -111,11 +119,17 @@ impl fmt::Display for Error {
  Self::FeerateEstimationUpdateFailed => {
  write!(f, "Failed to update fee rate estimates.")
  },
+ Self::FeerateEstimationUpdateTimeout => {
+ write!(f, "Updating fee rate estimates timed out.")
+ },
  Self::WalletOperationFailed => write!(f, "Failed to conduct wallet operation."),
+ Self::WalletOperationTimeout => write!(f, "A wallet operation timed out."),
  Self::OnchainTxSigningFailed => write!(f, "Failed to sign given transaction."),
  Self::MessageSigningFailed => write!(f, "Failed to sign given message."),
  Self::TxSyncFailed => write!(f, "Failed to sync transactions."),
+ Self::TxSyncTimeout => write!(f, "Syncing transactions timed out."),
  Self::GossipUpdateFailed => write!(f, "Failed to update gossip data."),
+ Self::GossipUpdateTimeout => write!(f, "Updating gossip data timed out."),
  Self::LiquidityRequestFailed => write!(f, "Failed to request inbound liquidity."),
  Self::InvalidAddress => write!(f, "The given address is invalid."),
  Self::InvalidSocketAddress => write!(f, "The given network address is invalid."),

src/fee_estimator.rs

@@ -1,3 +1,4 @@
+use crate::config::FEE_RATE_CACHE_UPDATE_TIMEOUT_SECS;
 use crate::logger::{log_error, log_trace, Logger};
 use crate::{Config, Error};
 
@@ -14,6 +15,7 @@ use bitcoin::Network;
 use std::collections::HashMap;
 use std::ops::Deref;
 use std::sync::{Arc, RwLock};
+use std::time::Duration;
 
 pub(crate) struct OnchainFeeEstimator<L: Deref>
 where
@@ -55,7 +57,21 @@ where
  ConfirmationTarget::OutputSpendingFee => 12,
  };
 
- let estimates = self.esplora_client.get_fee_estimates().await.map_err(|e| {
+ let estimates = tokio::time::timeout(
+ Duration::from_secs(FEE_RATE_CACHE_UPDATE_TIMEOUT_SECS),
+ self.esplora_client.get_fee_estimates(),
+ )
+ .await
+ .map_err(|e| {
+ log_error!(
+ self.logger,
+ "Updating fee rate estimates for {:?} timed out: {}",
+ target,
+ e
+ );
+ Error::FeerateEstimationUpdateTimeout
+ })?
+ .map_err(|e| {
  log_error!(
  self.logger,
  "Failed to retrieve fee rate estimates for {:?}: {}",

src/gossip.rs

@@ -1,3 +1,4 @@
+use crate::config::RGS_SYNC_TIMEOUT_SECS;
 use crate::logger::{log_trace, FilesystemLogger, Logger};
 use crate::types::{GossipSync, Graph, P2PGossipSync, RapidGossipSync};
 use crate::Error;
@@ -6,6 +7,7 @@ use lightning::routing::utxo::UtxoLookup;
 
 use std::sync::atomic::{AtomicU32, Ordering};
 use std::sync::Arc;
+use std::time::Duration;
 
 pub(crate) enum GossipSource {
 	P2PNetwork {
@@ -55,7 +57,17 @@ impl GossipSource {
  Self::RapidGossipSync { gossip_sync, server_url, latest_sync_timestamp, logger } => {
  let query_timestamp = latest_sync_timestamp.load(Ordering::Acquire);
  let query_url = format!("{}/{}", server_url, query_timestamp);
- let response = reqwest::get(query_url).await.map_err(|e| {
+
+ let response = tokio::time::timeout(
+ Duration::from_secs(RGS_SYNC_TIMEOUT_SECS),
+ reqwest::get(query_url),
+ )
+ .await
+ .map_err(|e| {
+ log_trace!(logger, "Retrieving RGS gossip update timed out: {}", e);
+ Error::GossipUpdateTimeout
+ })?
+ .map_err(|e| {
  log_trace!(logger, "Failed to retrieve RGS gossip update: {}", e);
  Error::GossipUpdateFailed
  })?;

Cargo.toml

@@ -64,7 +64,7 @@ bip39 = "2.0.0"
 
 rand = "0.8.5"
 chrono = { version = "0.4", default-features = false, features = ["clock"] }
-tokio = { version = "1", default-features = false, features = [ "rt-multi-thread", "time", "sync" ] }
+tokio = { version = "1.37", default-features = false, features = [ "rt-multi-thread", "time", "sync" ] }
 esplora-client = { version = "0.6", default-features = false }
 libc = "0.2"
 uniffi = { version = "0.26.0", features = ["build"], optional = true }

bindings/ldk_node.udl

@@ -165,11 +165,15 @@ enum NodeError {
 	"ChannelConfigUpdateFailed",
 	"PersistenceFailed",
 	"FeerateEstimationUpdateFailed",
+	"FeerateEstimationUpdateTimeout",
 	"WalletOperationFailed",
+	"WalletOperationTimeout",
 	"OnchainTxSigningFailed",
 	"MessageSigningFailed",
 	"TxSyncFailed",
+	"TxSyncTimeout",
 	"GossipUpdateFailed",
+	"GossipUpdateTimeout",
 	"LiquidityRequestFailed",
 	"InvalidAddress",
 	"InvalidSocketAddress",

src/builder.rs

@@ -1,6 +1,6 @@
 use crate::config::{
 	default_user_config, Config, BDK_CLIENT_CONCURRENCY, BDK_CLIENT_STOP_GAP,
-	DEFAULT_ESPLORA_SERVER_URL, WALLET_KEYS_SEED_LEN,
+DEFAULT_ESPLORA_CLIENT_TIMEOUT_SECS, DEFAULT_ESPLORA_SERVER_URL, WALLET_KEYS_SEED_LEN,
 };
 use crate::connection::ConnectionManager;
 use crate::event::EventQueue;
@@ -558,10 +558,15 @@ fn build_with_store_internal(
 
 	let (blockchain, tx_sync, tx_broadcaster, fee_estimator) = match chain_data_source_config {
  Some(ChainDataSourceConfig::Esplora(server_url)) => {
- let tx_sync = Arc::new(EsploraSyncClient::new(server_url.clone(), Arc::clone(&logger)));
- let blockchain =
- EsploraBlockchain::from_client(tx_sync.client().clone(), BDK_CLIENT_STOP_GAP)
- .with_concurrency(BDK_CLIENT_CONCURRENCY);
+ let mut client_builder = esplora_client::Builder::new(&server_url.clone());
+ client_builder = client_builder.timeout(DEFAULT_ESPLORA_CLIENT_TIMEOUT_SECS);
+ let esplora_client = client_builder.build_async().unwrap();
+ let tx_sync = Arc::new(EsploraSyncClient::from_client(
+ esplora_client.clone(),
+ Arc::clone(&logger),
+ ));
+ let blockchain = EsploraBlockchain::from_client(esplora_client, BDK_CLIENT_STOP_GAP)
+ .with_concurrency(BDK_CLIENT_CONCURRENCY);
  let tx_broadcaster = Arc::new(TransactionBroadcaster::new(
  tx_sync.client().clone(),
  Arc::clone(&logger),
@@ -959,6 +964,7 @@ fn build_with_store_internal(
 	};
 
 	let (stop_sender, _) = tokio::sync::watch::channel(());
+	let (event_handling_stopped_sender, _) = tokio::sync::watch::channel(());
 
 	let is_listening = Arc::new(AtomicBool::new(false));
 	let latest_wallet_sync_timestamp = Arc::new(RwLock::new(None));
@@ -971,6 +977,7 @@ fn build_with_store_internal(
 	Ok(Node {
  runtime,
  stop_sender,
+ event_handling_stopped_sender,
  config,
  wallet,
  tx_sync,

src/config.rs

@@ -28,6 +28,9 @@ pub(crate) const BDK_CLIENT_CONCURRENCY: u8 = 4;
 // The default Esplora server we're using.
 pub(crate) const DEFAULT_ESPLORA_SERVER_URL: &str = "https://blockstream.info/api";
 
+// The default Esplora client timeout we're using.
+pub(crate) const DEFAULT_ESPLORA_CLIENT_TIMEOUT_SECS: u64 = 10;
+
 // The timeout after which we abandon retrying failed payments.
 pub(crate) const LDK_PAYMENT_RETRY_TIMEOUT: Duration = Duration::from_secs(10);
 
@@ -46,6 +49,21 @@ pub(crate) const NODE_ANN_BCAST_INTERVAL: Duration = Duration::from_secs(60 * 60
 // The lower limit which we apply to any configured wallet sync intervals.
 pub(crate) const WALLET_SYNC_INTERVAL_MINIMUM_SECS: u64 = 10;
 
+// The timeout after which we abort a wallet syncing operation.
+pub(crate) const BDK_WALLET_SYNC_TIMEOUT_SECS: u64 = 90;
+
+// The timeout after which we abort a wallet syncing operation.
+pub(crate) const LDK_WALLET_SYNC_TIMEOUT_SECS: u64 = 30;
+
+// The timeout after which we abort a fee rate cache update operation.
+pub(crate) const FEE_RATE_CACHE_UPDATE_TIMEOUT_SECS: u64 = 5;
+
+// The timeout after which we abort a transaction broadcast operation.
+pub(crate) const TX_BROADCAST_TIMEOUT_SECS: u64 = 5;
+
+// The timeout after which we abort a RGS sync operation.
+pub(crate) const RGS_SYNC_TIMEOUT_SECS: u64 = 5;
+
 // The length in bytes of our wallets' keys seed.
 pub(crate) const WALLET_KEYS_SEED_LEN: usize = 64;
 

src/error.rs

@@ -33,16 +33,24 @@ pub enum Error {
 	PersistenceFailed,
 	/// A fee rate estimation update failed.
 	FeerateEstimationUpdateFailed,
+	/// A fee rate estimation update timed out.
+	FeerateEstimationUpdateTimeout,
 	/// A wallet operation failed.
 	WalletOperationFailed,
+	/// A wallet operation timed out.
+	WalletOperationTimeout,
 	/// A signing operation for transaction failed.
 	OnchainTxSigningFailed,
 	/// A signing operation for message failed.
 	MessageSigningFailed,
 	/// A transaction sync operation failed.
 	TxSyncFailed,
+	/// A transaction sync operation timed out.
+	TxSyncTimeout,
 	/// A gossip updating operation failed.
 	GossipUpdateFailed,
+	/// A gossip updating operation timed out.
+	GossipUpdateTimeout,
 	/// A liquidity request operation failed.
 	LiquidityRequestFailed,
 	/// The given address is invalid.
@@ -111,11 +119,17 @@ impl fmt::Display for Error {
  Self::FeerateEstimationUpdateFailed => {
  write!(f, "Failed to update fee rate estimates.")
  },
+ Self::FeerateEstimationUpdateTimeout => {
+ write!(f, "Updating fee rate estimates timed out.")
+ },
  Self::WalletOperationFailed => write!(f, "Failed to conduct wallet operation."),
+ Self::WalletOperationTimeout => write!(f, "A wallet operation timed out."),
  Self::OnchainTxSigningFailed => write!(f, "Failed to sign given transaction."),
  Self::MessageSigningFailed => write!(f, "Failed to sign given message."),
  Self::TxSyncFailed => write!(f, "Failed to sync transactions."),
+ Self::TxSyncTimeout => write!(f, "Syncing transactions timed out."),
  Self::GossipUpdateFailed => write!(f, "Failed to update gossip data."),
+ Self::GossipUpdateTimeout => write!(f, "Updating gossip data timed out."),
  Self::LiquidityRequestFailed => write!(f, "Failed to request inbound liquidity."),
  Self::InvalidAddress => write!(f, "The given address is invalid."),
  Self::InvalidSocketAddress => write!(f, "The given network address is invalid."),

src/fee_estimator.rs

@@ -1,3 +1,4 @@
+use crate::config::FEE_RATE_CACHE_UPDATE_TIMEOUT_SECS;
 use crate::logger::{log_error, log_trace, Logger};
 use crate::{Config, Error};
 
@@ -14,6 +15,7 @@ use bitcoin::Network;
 use std::collections::HashMap;
 use std::ops::Deref;
 use std::sync::{Arc, RwLock};
+use std::time::Duration;
 
 pub(crate) struct OnchainFeeEstimator<L: Deref>
 where
@@ -55,7 +57,21 @@ where
  ConfirmationTarget::OutputSpendingFee => 12,
  };
 
- let estimates = self.esplora_client.get_fee_estimates().await.map_err(|e| {
+ let estimates = tokio::time::timeout(
+ Duration::from_secs(FEE_RATE_CACHE_UPDATE_TIMEOUT_SECS),
+ self.esplora_client.get_fee_estimates(),
+ )
+ .await
+ .map_err(|e| {
+ log_error!(
+ self.logger,
+ "Updating fee rate estimates for {:?} timed out: {}",
+ target,
+ e
+ );
+ Error::FeerateEstimationUpdateTimeout
+ })?
+ .map_err(|e| {
  log_error!(
  self.logger,
  "Failed to retrieve fee rate estimates for {:?}: {}",

src/gossip.rs

@@ -1,3 +1,4 @@
+use crate::config::RGS_SYNC_TIMEOUT_SECS;
 use crate::logger::{log_trace, FilesystemLogger, Logger};
 use crate::types::{GossipSync, Graph, P2PGossipSync, RapidGossipSync};
 use crate::Error;
@@ -6,6 +7,7 @@ use lightning::routing::utxo::UtxoLookup;
 
 use std::sync::atomic::{AtomicU32, Ordering};
 use std::sync::Arc;
+use std::time::Duration;
 
 pub(crate) enum GossipSource {
 	P2PNetwork {
@@ -55,7 +57,17 @@ impl GossipSource {
  Self::RapidGossipSync { gossip_sync, server_url, latest_sync_timestamp, logger } => {
  let query_timestamp = latest_sync_timestamp.load(Ordering::Acquire);
  let query_url = format!("{}/{}", server_url, query_timestamp);
- let response = reqwest::get(query_url).await.map_err(|e| {
+
+ let response = tokio::time::timeout(
+ Duration::from_secs(RGS_SYNC_TIMEOUT_SECS),
+ reqwest::get(query_url),
+ )
+ .await
+ .map_err(|e| {
+ log_trace!(logger, "Retrieving RGS gossip update timed out: {}", e);
+ Error::GossipUpdateTimeout
+ })?
+ .map_err(|e| {
  log_trace!(logger, "Failed to retrieve RGS gossip update: {}", e);
  Error::GossipUpdateFailed
  })?;

Cargo.toml

@@ -64,7 +64,7 @@ bip39 = "2.0.0"
 
 rand = "0.8.5"
 chrono = { version = "0.4", default-features = false, features = ["clock"] }
-tokio = { version = "1", default-features = false, features = [ "rt-multi-thread", "time", "sync" ] }
+tokio = { version = "1.37", default-features = false, features = [ "rt-multi-thread", "time", "sync" ] }
 esplora-client = { version = "0.6", default-features = false }
 libc = "0.2"
 uniffi = { version = "0.26.0", features = ["build"], optional = true }

bindings/ldk_node.udl

@@ -165,11 +165,15 @@ enum NodeError {
 	"ChannelConfigUpdateFailed",
 	"PersistenceFailed",
 	"FeerateEstimationUpdateFailed",
+	"FeerateEstimationUpdateTimeout",
 	"WalletOperationFailed",
+	"WalletOperationTimeout",
 	"OnchainTxSigningFailed",
 	"MessageSigningFailed",
 	"TxSyncFailed",
+	"TxSyncTimeout",
 	"GossipUpdateFailed",
+	"GossipUpdateTimeout",
 	"LiquidityRequestFailed",
 	"InvalidAddress",
 	"InvalidSocketAddress",

src/builder.rs

@@ -1,6 +1,6 @@
 use crate::config::{
 	default_user_config, Config, BDK_CLIENT_CONCURRENCY, BDK_CLIENT_STOP_GAP,
-	DEFAULT_ESPLORA_SERVER_URL, WALLET_KEYS_SEED_LEN,
+DEFAULT_ESPLORA_CLIENT_TIMEOUT_SECS, DEFAULT_ESPLORA_SERVER_URL, WALLET_KEYS_SEED_LEN,
 };
 use crate::connection::ConnectionManager;
 use crate::event::EventQueue;
@@ -558,10 +558,15 @@ fn build_with_store_internal(
 
 	let (blockchain, tx_sync, tx_broadcaster, fee_estimator) = match chain_data_source_config {
  Some(ChainDataSourceConfig::Esplora(server_url)) => {
- let tx_sync = Arc::new(EsploraSyncClient::new(server_url.clone(), Arc::clone(&logger)));
- let blockchain =
- EsploraBlockchain::from_client(tx_sync.client().clone(), BDK_CLIENT_STOP_GAP)
- .with_concurrency(BDK_CLIENT_CONCURRENCY);
+ let mut client_builder = esplora_client::Builder::new(&server_url.clone());
+ client_builder = client_builder.timeout(DEFAULT_ESPLORA_CLIENT_TIMEOUT_SECS);
+ let esplora_client = client_builder.build_async().unwrap();
+ let tx_sync = Arc::new(EsploraSyncClient::from_client(
+ esplora_client.clone(),
+ Arc::clone(&logger),
+ ));
+ let blockchain = EsploraBlockchain::from_client(esplora_client, BDK_CLIENT_STOP_GAP)
+ .with_concurrency(BDK_CLIENT_CONCURRENCY);
  let tx_broadcaster = Arc::new(TransactionBroadcaster::new(
  tx_sync.client().clone(),
  Arc::clone(&logger),
@@ -959,6 +964,7 @@ fn build_with_store_internal(
 	};
 
 	let (stop_sender, _) = tokio::sync::watch::channel(());
+	let (event_handling_stopped_sender, _) = tokio::sync::watch::channel(());
 
 	let is_listening = Arc::new(AtomicBool::new(false));
 	let latest_wallet_sync_timestamp = Arc::new(RwLock::new(None));
@@ -971,6 +977,7 @@ fn build_with_store_internal(
 	Ok(Node {
  runtime,
  stop_sender,
+ event_handling_stopped_sender,
  config,
  wallet,
  tx_sync,

src/config.rs

@@ -28,6 +28,9 @@ pub(crate) const BDK_CLIENT_CONCURRENCY: u8 = 4;
 // The default Esplora server we're using.
 pub(crate) const DEFAULT_ESPLORA_SERVER_URL: &str = "https://blockstream.info/api";
 
+// The default Esplora client timeout we're using.
+pub(crate) const DEFAULT_ESPLORA_CLIENT_TIMEOUT_SECS: u64 = 10;
+
 // The timeout after which we abandon retrying failed payments.
 pub(crate) const LDK_PAYMENT_RETRY_TIMEOUT: Duration = Duration::from_secs(10);
 
@@ -46,6 +49,21 @@ pub(crate) const NODE_ANN_BCAST_INTERVAL: Duration = Duration::from_secs(60 * 60
 // The lower limit which we apply to any configured wallet sync intervals.
 pub(crate) const WALLET_SYNC_INTERVAL_MINIMUM_SECS: u64 = 10;
 
+// The timeout after which we abort a wallet syncing operation.
+pub(crate) const BDK_WALLET_SYNC_TIMEOUT_SECS: u64 = 90;
+
+// The timeout after which we abort a wallet syncing operation.
+pub(crate) const LDK_WALLET_SYNC_TIMEOUT_SECS: u64 = 30;
+
+// The timeout after which we abort a fee rate cache update operation.
+pub(crate) const FEE_RATE_CACHE_UPDATE_TIMEOUT_SECS: u64 = 5;
+
+// The timeout after which we abort a transaction broadcast operation.
+pub(crate) const TX_BROADCAST_TIMEOUT_SECS: u64 = 5;
+
+// The timeout after which we abort a RGS sync operation.
+pub(crate) const RGS_SYNC_TIMEOUT_SECS: u64 = 5;
+
 // The length in bytes of our wallets' keys seed.
 pub(crate) const WALLET_KEYS_SEED_LEN: usize = 64;
 

src/error.rs

@@ -33,16 +33,24 @@ pub enum Error {
 	PersistenceFailed,
 	/// A fee rate estimation update failed.
 	FeerateEstimationUpdateFailed,
+	/// A fee rate estimation update timed out.
+	FeerateEstimationUpdateTimeout,
 	/// A wallet operation failed.
 	WalletOperationFailed,
+	/// A wallet operation timed out.
+	WalletOperationTimeout,
 	/// A signing operation for transaction failed.
 	OnchainTxSigningFailed,
 	/// A signing operation for message failed.
 	MessageSigningFailed,
 	/// A transaction sync operation failed.
 	TxSyncFailed,
+	/// A transaction sync operation timed out.
+	TxSyncTimeout,
 	/// A gossip updating operation failed.
 	GossipUpdateFailed,
+	/// A gossip updating operation timed out.
+	GossipUpdateTimeout,
 	/// A liquidity request operation failed.
 	LiquidityRequestFailed,
 	/// The given address is invalid.
@@ -111,11 +119,17 @@ impl fmt::Display for Error {
  Self::FeerateEstimationUpdateFailed => {
  write!(f, "Failed to update fee rate estimates.")
  },
+ Self::FeerateEstimationUpdateTimeout => {
+ write!(f, "Updating fee rate estimates timed out.")
+ },
  Self::WalletOperationFailed => write!(f, "Failed to conduct wallet operation."),
+ Self::WalletOperationTimeout => write!(f, "A wallet operation timed out."),
  Self::OnchainTxSigningFailed => write!(f, "Failed to sign given transaction."),
  Self::MessageSigningFailed => write!(f, "Failed to sign given message."),
  Self::TxSyncFailed => write!(f, "Failed to sync transactions."),
+ Self::TxSyncTimeout => write!(f, "Syncing transactions timed out."),
  Self::GossipUpdateFailed => write!(f, "Failed to update gossip data."),
+ Self::GossipUpdateTimeout => write!(f, "Updating gossip data timed out."),
  Self::LiquidityRequestFailed => write!(f, "Failed to request inbound liquidity."),
  Self::InvalidAddress => write!(f, "The given address is invalid."),
  Self::InvalidSocketAddress => write!(f, "The given network address is invalid."),

src/fee_estimator.rs

@@ -1,3 +1,4 @@
+use crate::config::FEE_RATE_CACHE_UPDATE_TIMEOUT_SECS;
 use crate::logger::{log_error, log_trace, Logger};
 use crate::{Config, Error};
 
@@ -14,6 +15,7 @@ use bitcoin::Network;
 use std::collections::HashMap;
 use std::ops::Deref;
 use std::sync::{Arc, RwLock};
+use std::time::Duration;
 
 pub(crate) struct OnchainFeeEstimator<L: Deref>
 where
@@ -55,7 +57,21 @@ where
  ConfirmationTarget::OutputSpendingFee => 12,
  };
 
- let estimates = self.esplora_client.get_fee_estimates().await.map_err(|e| {
+ let estimates = tokio::time::timeout(
+ Duration::from_secs(FEE_RATE_CACHE_UPDATE_TIMEOUT_SECS),
+ self.esplora_client.get_fee_estimates(),
+ )
+ .await
+ .map_err(|e| {
+ log_error!(
+ self.logger,
+ "Updating fee rate estimates for {:?} timed out: {}",
+ target,
+ e
+ );
+ Error::FeerateEstimationUpdateTimeout
+ })?
+ .map_err(|e| {
  log_error!(
  self.logger,
  "Failed to retrieve fee rate estimates for {:?}: {}",

src/gossip.rs

@@ -1,3 +1,4 @@
+use crate::config::RGS_SYNC_TIMEOUT_SECS;
 use crate::logger::{log_trace, FilesystemLogger, Logger};
 use crate::types::{GossipSync, Graph, P2PGossipSync, RapidGossipSync};
 use crate::Error;
@@ -6,6 +7,7 @@ use lightning::routing::utxo::UtxoLookup;
 
 use std::sync::atomic::{AtomicU32, Ordering};
 use std::sync::Arc;
+use std::time::Duration;
 
 pub(crate) enum GossipSource {
 	P2PNetwork {
@@ -55,7 +57,17 @@ impl GossipSource {
  Self::RapidGossipSync { gossip_sync, server_url, latest_sync_timestamp, logger } => {
  let query_timestamp = latest_sync_timestamp.load(Ordering::Acquire);
  let query_url = format!("{}/{}", server_url, query_timestamp);
- let response = reqwest::get(query_url).await.map_err(|e| {
+
+ let response = tokio::time::timeout(
+ Duration::from_secs(RGS_SYNC_TIMEOUT_SECS),
+ reqwest::get(query_url),
+ )
+ .await
+ .map_err(|e| {
+ log_trace!(logger, "Retrieving RGS gossip update timed out: {}", e);
+ Error::GossipUpdateTimeout
+ })?
+ .map_err(|e| {
  log_trace!(logger, "Failed to retrieve RGS gossip update: {}", e);
  Error::GossipUpdateFailed
  })?;

Cargo.toml

@@ -64,7 +64,7 @@ bip39 = "2.0.0"
 
 rand = "0.8.5"
 chrono = { version = "0.4", default-features = false, features = ["clock"] }
-tokio = { version = "1", default-features = false, features = [ "rt-multi-thread", "time", "sync" ] }
+tokio = { version = "1.37", default-features = false, features = [ "rt-multi-thread", "time", "sync" ] }
 esplora-client = { version = "0.6", default-features = false }
 libc = "0.2"
 uniffi = { version = "0.26.0", features = ["build"], optional = true }

bindings/ldk_node.udl

@@ -165,11 +165,15 @@ enum NodeError {
 	"ChannelConfigUpdateFailed",
 	"PersistenceFailed",
 	"FeerateEstimationUpdateFailed",
+	"FeerateEstimationUpdateTimeout",
 	"WalletOperationFailed",
+	"WalletOperationTimeout",
 	"OnchainTxSigningFailed",
 	"MessageSigningFailed",
 	"TxSyncFailed",
+	"TxSyncTimeout",
 	"GossipUpdateFailed",
+	"GossipUpdateTimeout",
 	"LiquidityRequestFailed",
 	"InvalidAddress",
 	"InvalidSocketAddress",

src/builder.rs

@@ -1,6 +1,6 @@
 use crate::config::{
 	default_user_config, Config, BDK_CLIENT_CONCURRENCY, BDK_CLIENT_STOP_GAP,
-	DEFAULT_ESPLORA_SERVER_URL, WALLET_KEYS_SEED_LEN,
+DEFAULT_ESPLORA_CLIENT_TIMEOUT_SECS, DEFAULT_ESPLORA_SERVER_URL, WALLET_KEYS_SEED_LEN,
 };
 use crate::connection::ConnectionManager;
 use crate::event::EventQueue;
@@ -558,10 +558,15 @@ fn build_with_store_internal(
 
 	let (blockchain, tx_sync, tx_broadcaster, fee_estimator) = match chain_data_source_config {
  Some(ChainDataSourceConfig::Esplora(server_url)) => {
- let tx_sync = Arc::new(EsploraSyncClient::new(server_url.clone(), Arc::clone(&logger)));
- let blockchain =
- EsploraBlockchain::from_client(tx_sync.client().clone(), BDK_CLIENT_STOP_GAP)
- .with_concurrency(BDK_CLIENT_CONCURRENCY);
+ let mut client_builder = esplora_client::Builder::new(&server_url.clone());
+ client_builder = client_builder.timeout(DEFAULT_ESPLORA_CLIENT_TIMEOUT_SECS);
+ let esplora_client = client_builder.build_async().unwrap();
+ let tx_sync = Arc::new(EsploraSyncClient::from_client(
+ esplora_client.clone(),
+ Arc::clone(&logger),
+ ));
+ let blockchain = EsploraBlockchain::from_client(esplora_client, BDK_CLIENT_STOP_GAP)
+ .with_concurrency(BDK_CLIENT_CONCURRENCY);
  let tx_broadcaster = Arc::new(TransactionBroadcaster::new(
  tx_sync.client().clone(),
  Arc::clone(&logger),
@@ -959,6 +964,7 @@ fn build_with_store_internal(
 	};
 
 	let (stop_sender, _) = tokio::sync::watch::channel(());
+	let (event_handling_stopped_sender, _) = tokio::sync::watch::channel(());
 
 	let is_listening = Arc::new(AtomicBool::new(false));
 	let latest_wallet_sync_timestamp = Arc::new(RwLock::new(None));
@@ -971,6 +977,7 @@ fn build_with_store_internal(
 	Ok(Node {
  runtime,
  stop_sender,
+ event_handling_stopped_sender,
  config,
  wallet,
  tx_sync,

src/config.rs

@@ -28,6 +28,9 @@ pub(crate) const BDK_CLIENT_CONCURRENCY: u8 = 4;
 // The default Esplora server we're using.
 pub(crate) const DEFAULT_ESPLORA_SERVER_URL: &str = "https://blockstream.info/api";
 
+// The default Esplora client timeout we're using.
+pub(crate) const DEFAULT_ESPLORA_CLIENT_TIMEOUT_SECS: u64 = 10;
+
 // The timeout after which we abandon retrying failed payments.
 pub(crate) const LDK_PAYMENT_RETRY_TIMEOUT: Duration = Duration::from_secs(10);
 
@@ -46,6 +49,21 @@ pub(crate) const NODE_ANN_BCAST_INTERVAL: Duration = Duration::from_secs(60 * 60
 // The lower limit which we apply to any configured wallet sync intervals.
 pub(crate) const WALLET_SYNC_INTERVAL_MINIMUM_SECS: u64 = 10;
 
+// The timeout after which we abort a wallet syncing operation.
+pub(crate) const BDK_WALLET_SYNC_TIMEOUT_SECS: u64 = 90;
+
+// The timeout after which we abort a wallet syncing operation.
+pub(crate) const LDK_WALLET_SYNC_TIMEOUT_SECS: u64 = 30;
+
+// The timeout after which we abort a fee rate cache update operation.
+pub(crate) const FEE_RATE_CACHE_UPDATE_TIMEOUT_SECS: u64 = 5;
+
+// The timeout after which we abort a transaction broadcast operation.
+pub(crate) const TX_BROADCAST_TIMEOUT_SECS: u64 = 5;
+
+// The timeout after which we abort a RGS sync operation.
+pub(crate) const RGS_SYNC_TIMEOUT_SECS: u64 = 5;
+
 // The length in bytes of our wallets' keys seed.
 pub(crate) const WALLET_KEYS_SEED_LEN: usize = 64;
 

src/error.rs

@@ -33,16 +33,24 @@ pub enum Error {
 	PersistenceFailed,
 	/// A fee rate estimation update failed.
 	FeerateEstimationUpdateFailed,
+	/// A fee rate estimation update timed out.
+	FeerateEstimationUpdateTimeout,
 	/// A wallet operation failed.
 	WalletOperationFailed,
+	/// A wallet operation timed out.
+	WalletOperationTimeout,
 	/// A signing operation for transaction failed.
 	OnchainTxSigningFailed,
 	/// A signing operation for message failed.
 	MessageSigningFailed,
 	/// A transaction sync operation failed.
 	TxSyncFailed,
+	/// A transaction sync operation timed out.
+	TxSyncTimeout,
 	/// A gossip updating operation failed.
 	GossipUpdateFailed,
+	/// A gossip updating operation timed out.
+	GossipUpdateTimeout,
 	/// A liquidity request operation failed.
 	LiquidityRequestFailed,
 	/// The given address is invalid.
@@ -111,11 +119,17 @@ impl fmt::Display for Error {
  Self::FeerateEstimationUpdateFailed => {
  write!(f, "Failed to update fee rate estimates.")
  },
+ Self::FeerateEstimationUpdateTimeout => {
+ write!(f, "Updating fee rate estimates timed out.")
+ },
  Self::WalletOperationFailed => write!(f, "Failed to conduct wallet operation."),
+ Self::WalletOperationTimeout => write!(f, "A wallet operation timed out."),
  Self::OnchainTxSigningFailed => write!(f, "Failed to sign given transaction."),
  Self::MessageSigningFailed => write!(f, "Failed to sign given message."),
  Self::TxSyncFailed => write!(f, "Failed to sync transactions."),
+ Self::TxSyncTimeout => write!(f, "Syncing transactions timed out."),
  Self::GossipUpdateFailed => write!(f, "Failed to update gossip data."),
+ Self::GossipUpdateTimeout => write!(f, "Updating gossip data timed out."),
  Self::LiquidityRequestFailed => write!(f, "Failed to request inbound liquidity."),
  Self::InvalidAddress => write!(f, "The given address is invalid."),
  Self::InvalidSocketAddress => write!(f, "The given network address is invalid."),

src/fee_estimator.rs

@@ -1,3 +1,4 @@
+use crate::config::FEE_RATE_CACHE_UPDATE_TIMEOUT_SECS;
 use crate::logger::{log_error, log_trace, Logger};
 use crate::{Config, Error};
 
@@ -14,6 +15,7 @@ use bitcoin::Network;
 use std::collections::HashMap;
 use std::ops::Deref;
 use std::sync::{Arc, RwLock};
+use std::time::Duration;
 
 pub(crate) struct OnchainFeeEstimator<L: Deref>
 where
@@ -55,7 +57,21 @@ where
  ConfirmationTarget::OutputSpendingFee => 12,
  };
 
- let estimates = self.esplora_client.get_fee_estimates().await.map_err(|e| {
+ let estimates = tokio::time::timeout(
+ Duration::from_secs(FEE_RATE_CACHE_UPDATE_TIMEOUT_SECS),
+ self.esplora_client.get_fee_estimates(),
+ )
+ .await
+ .map_err(|e| {
+ log_error!(
+ self.logger,
+ "Updating fee rate estimates for {:?} timed out: {}",
+ target,
+ e
+ );
+ Error::FeerateEstimationUpdateTimeout
+ })?
+ .map_err(|e| {
  log_error!(
  self.logger,
  "Failed to retrieve fee rate estimates for {:?}: {}",

src/gossip.rs

@@ -1,3 +1,4 @@
+use crate::config::RGS_SYNC_TIMEOUT_SECS;
 use crate::logger::{log_trace, FilesystemLogger, Logger};
 use crate::types::{GossipSync, Graph, P2PGossipSync, RapidGossipSync};
 use crate::Error;
@@ -6,6 +7,7 @@ use lightning::routing::utxo::UtxoLookup;
 
 use std::sync::atomic::{AtomicU32, Ordering};
 use std::sync::Arc;
+use std::time::Duration;
 
 pub(crate) enum GossipSource {
 	P2PNetwork {
@@ -55,7 +57,17 @@ impl GossipSource {
  Self::RapidGossipSync { gossip_sync, server_url, latest_sync_timestamp, logger } => {
  let query_timestamp = latest_sync_timestamp.load(Ordering::Acquire);
  let query_url = format!("{}/{}", server_url, query_timestamp);
- let response = reqwest::get(query_url).await.map_err(|e| {
+
+ let response = tokio::time::timeout(
+ Duration::from_secs(RGS_SYNC_TIMEOUT_SECS),
+ reqwest::get(query_url),
+ )
+ .await
+ .map_err(|e| {
+ log_trace!(logger, "Retrieving RGS gossip update timed out: {}", e);
+ Error::GossipUpdateTimeout
+ })?
+ .map_err(|e| {
  log_trace!(logger, "Failed to retrieve RGS gossip update: {}", e);
  Error::GossipUpdateFailed
  })?;

Cargo.toml

@@ -64,7 +64,7 @@ bip39 = "2.0.0"
 
 rand = "0.8.5"
 chrono = { version = "0.4", default-features = false, features = ["clock"] }
-tokio = { version = "1", default-features = false, features = [ "rt-multi-thread", "time", "sync" ] }
+tokio = { version = "1.37", default-features = false, features = [ "rt-multi-thread", "time", "sync" ] }
 esplora-client = { version = "0.6", default-features = false }
 libc = "0.2"
 uniffi = { version = "0.26.0", features = ["build"], optional = true }

bindings/ldk_node.udl

@@ -165,11 +165,15 @@ enum NodeError {
 	"ChannelConfigUpdateFailed",
 	"PersistenceFailed",
 	"FeerateEstimationUpdateFailed",
+	"FeerateEstimationUpdateTimeout",
 	"WalletOperationFailed",
+	"WalletOperationTimeout",
 	"OnchainTxSigningFailed",
 	"MessageSigningFailed",
 	"TxSyncFailed",
+	"TxSyncTimeout",
 	"GossipUpdateFailed",
+	"GossipUpdateTimeout",
 	"LiquidityRequestFailed",
 	"InvalidAddress",
 	"InvalidSocketAddress",

src/builder.rs

@@ -1,6 +1,6 @@
 use crate::config::{
 	default_user_config, Config, BDK_CLIENT_CONCURRENCY, BDK_CLIENT_STOP_GAP,
-	DEFAULT_ESPLORA_SERVER_URL, WALLET_KEYS_SEED_LEN,
+DEFAULT_ESPLORA_CLIENT_TIMEOUT_SECS, DEFAULT_ESPLORA_SERVER_URL, WALLET_KEYS_SEED_LEN,
 };
 use crate::connection::ConnectionManager;
 use crate::event::EventQueue;
@@ -558,10 +558,15 @@ fn build_with_store_internal(
 
 	let (blockchain, tx_sync, tx_broadcaster, fee_estimator) = match chain_data_source_config {
  Some(ChainDataSourceConfig::Esplora(server_url)) => {
- let tx_sync = Arc::new(EsploraSyncClient::new(server_url.clone(), Arc::clone(&logger)));
- let blockchain =
- EsploraBlockchain::from_client(tx_sync.client().clone(), BDK_CLIENT_STOP_GAP)
- .with_concurrency(BDK_CLIENT_CONCURRENCY);
+ let mut client_builder = esplora_client::Builder::new(&server_url.clone());
+ client_builder = client_builder.timeout(DEFAULT_ESPLORA_CLIENT_TIMEOUT_SECS);
+ let esplora_client = client_builder.build_async().unwrap();
+ let tx_sync = Arc::new(EsploraSyncClient::from_client(
+ esplora_client.clone(),
+ Arc::clone(&logger),
+ ));
+ let blockchain = EsploraBlockchain::from_client(esplora_client, BDK_CLIENT_STOP_GAP)
+ .with_concurrency(BDK_CLIENT_CONCURRENCY);
  let tx_broadcaster = Arc::new(TransactionBroadcaster::new(
  tx_sync.client().clone(),
  Arc::clone(&logger),
@@ -959,6 +964,7 @@ fn build_with_store_internal(
 	};
 
 	let (stop_sender, _) = tokio::sync::watch::channel(());
+	let (event_handling_stopped_sender, _) = tokio::sync::watch::channel(());
 
 	let is_listening = Arc::new(AtomicBool::new(false));
 	let latest_wallet_sync_timestamp = Arc::new(RwLock::new(None));
@@ -971,6 +977,7 @@ fn build_with_store_internal(
 	Ok(Node {
  runtime,
  stop_sender,
+ event_handling_stopped_sender,
  config,
  wallet,
  tx_sync,

src/config.rs

@@ -28,6 +28,9 @@ pub(crate) const BDK_CLIENT_CONCURRENCY: u8 = 4;
 // The default Esplora server we're using.
 pub(crate) const DEFAULT_ESPLORA_SERVER_URL: &str = "https://blockstream.info/api";
 
+// The default Esplora client timeout we're using.
+pub(crate) const DEFAULT_ESPLORA_CLIENT_TIMEOUT_SECS: u64 = 10;
+
 // The timeout after which we abandon retrying failed payments.
 pub(crate) const LDK_PAYMENT_RETRY_TIMEOUT: Duration = Duration::from_secs(10);
 
@@ -46,6 +49,21 @@ pub(crate) const NODE_ANN_BCAST_INTERVAL: Duration = Duration::from_secs(60 * 60
 // The lower limit which we apply to any configured wallet sync intervals.
 pub(crate) const WALLET_SYNC_INTERVAL_MINIMUM_SECS: u64 = 10;
 
+// The timeout after which we abort a wallet syncing operation.
+pub(crate) const BDK_WALLET_SYNC_TIMEOUT_SECS: u64 = 90;
+
+// The timeout after which we abort a wallet syncing operation.
+pub(crate) const LDK_WALLET_SYNC_TIMEOUT_SECS: u64 = 30;
+
+// The timeout after which we abort a fee rate cache update operation.
+pub(crate) const FEE_RATE_CACHE_UPDATE_TIMEOUT_SECS: u64 = 5;
+
+// The timeout after which we abort a transaction broadcast operation.
+pub(crate) const TX_BROADCAST_TIMEOUT_SECS: u64 = 5;
+
+// The timeout after which we abort a RGS sync operation.
+pub(crate) const RGS_SYNC_TIMEOUT_SECS: u64 = 5;
+
 // The length in bytes of our wallets' keys seed.
 pub(crate) const WALLET_KEYS_SEED_LEN: usize = 64;
 

src/error.rs

@@ -33,16 +33,24 @@ pub enum Error {
 	PersistenceFailed,
 	/// A fee rate estimation update failed.
 	FeerateEstimationUpdateFailed,
+	/// A fee rate estimation update timed out.
+	FeerateEstimationUpdateTimeout,
 	/// A wallet operation failed.
 	WalletOperationFailed,
+	/// A wallet operation timed out.
+	WalletOperationTimeout,
 	/// A signing operation for transaction failed.
 	OnchainTxSigningFailed,
 	/// A signing operation for message failed.
 	MessageSigningFailed,
 	/// A transaction sync operation failed.
 	TxSyncFailed,
+	/// A transaction sync operation timed out.
+	TxSyncTimeout,
 	/// A gossip updating operation failed.
 	GossipUpdateFailed,
+	/// A gossip updating operation timed out.
+	GossipUpdateTimeout,
 	/// A liquidity request operation failed.
 	LiquidityRequestFailed,
 	/// The given address is invalid.
@@ -111,11 +119,17 @@ impl fmt::Display for Error {
  Self::FeerateEstimationUpdateFailed => {
  write!(f, "Failed to update fee rate estimates.")
  },
+ Self::FeerateEstimationUpdateTimeout => {
+ write!(f, "Updating fee rate estimates timed out.")
+ },
  Self::WalletOperationFailed => write!(f, "Failed to conduct wallet operation."),
+ Self::WalletOperationTimeout => write!(f, "A wallet operation timed out."),
  Self::OnchainTxSigningFailed => write!(f, "Failed to sign given transaction."),
  Self::MessageSigningFailed => write!(f, "Failed to sign given message."),
  Self::TxSyncFailed => write!(f, "Failed to sync transactions."),
+ Self::TxSyncTimeout => write!(f, "Syncing transactions timed out."),
  Self::GossipUpdateFailed => write!(f, "Failed to update gossip data."),
+ Self::GossipUpdateTimeout => write!(f, "Updating gossip data timed out."),
  Self::LiquidityRequestFailed => write!(f, "Failed to request inbound liquidity."),
  Self::InvalidAddress => write!(f, "The given address is invalid."),
  Self::InvalidSocketAddress => write!(f, "The given network address is invalid."),

src/fee_estimator.rs

@@ -1,3 +1,4 @@
+use crate::config::FEE_RATE_CACHE_UPDATE_TIMEOUT_SECS;
 use crate::logger::{log_error, log_trace, Logger};
 use crate::{Config, Error};
 
@@ -14,6 +15,7 @@ use bitcoin::Network;
 use std::collections::HashMap;
 use std::ops::Deref;
 use std::sync::{Arc, RwLock};
+use std::time::Duration;
 
 pub(crate) struct OnchainFeeEstimator<L: Deref>
 where
@@ -55,7 +57,21 @@ where
  ConfirmationTarget::OutputSpendingFee => 12,
  };
 
- let estimates = self.esplora_client.get_fee_estimates().await.map_err(|e| {
+ let estimates = tokio::time::timeout(
+ Duration::from_secs(FEE_RATE_CACHE_UPDATE_TIMEOUT_SECS),
+ self.esplora_client.get_fee_estimates(),
+ )
+ .await
+ .map_err(|e| {
+ log_error!(
+ self.logger,
+ "Updating fee rate estimates for {:?} timed out: {}",
+ target,
+ e
+ );
+ Error::FeerateEstimationUpdateTimeout
+ })?
+ .map_err(|e| {
  log_error!(
  self.logger,
  "Failed to retrieve fee rate estimates for {:?}: {}",

src/gossip.rs

@@ -1,3 +1,4 @@
+use crate::config::RGS_SYNC_TIMEOUT_SECS;
 use crate::logger::{log_trace, FilesystemLogger, Logger};
 use crate::types::{GossipSync, Graph, P2PGossipSync, RapidGossipSync};
 use crate::Error;
@@ -6,6 +7,7 @@ use lightning::routing::utxo::UtxoLookup;
 
 use std::sync::atomic::{AtomicU32, Ordering};
 use std::sync::Arc;
+use std::time::Duration;
 
 pub(crate) enum GossipSource {
 	P2PNetwork {
@@ -55,7 +57,17 @@ impl GossipSource {
  Self::RapidGossipSync { gossip_sync, server_url, latest_sync_timestamp, logger } => {
  let query_timestamp = latest_sync_timestamp.load(Ordering::Acquire);
  let query_url = format!("{}/{}", server_url, query_timestamp);
- let response = reqwest::get(query_url).await.map_err(|e| {
+
+ let response = tokio::time::timeout(
+ Duration::from_secs(RGS_SYNC_TIMEOUT_SECS),
+ reqwest::get(query_url),
+ )
+ .await
+ .map_err(|e| {
+ log_trace!(logger, "Retrieving RGS gossip update timed out: {}", e);
+ Error::GossipUpdateTimeout
+ })?
+ .map_err(|e| {
  log_trace!(logger, "Failed to retrieve RGS gossip update: {}", e);
  Error::GossipUpdateFailed
  })?;