Move broadcast queue processing to ChainSource

Author: tnull

src/builder.rs

@@ -5,7 +5,7 @@
 // http://opensource.org/licenses/MIT>, at your option. You may not use this file except in
 // accordance with one or both of these licenses.
 
-use crate::chain::{ChainSource, DEFAULT_ESPLORA_CLIENT_TIMEOUT_SECS, DEFAULT_ESPLORA_SERVER_URL};
+use crate::chain::{ChainSource, DEFAULT_ESPLORA_SERVER_URL};
 use crate::config::{default_user_config, Config, WALLET_KEYS_SEED_LEN};
 
 use crate::connection::ConnectionManager;
@@ -583,60 +583,36 @@ fn build_with_store_internal(
 			})?,
 	};
 
-	let (wallet, chain_source, tx_broadcaster, fee_estimator) = match chain_data_source_config {
-		Some(ChainDataSourceConfig::Esplora(server_url)) => {
-			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_broadcaster =
-				Arc::new(TransactionBroadcaster::new(esplora_client.clone(), Arc::clone(&logger)));
-			let fee_estimator = Arc::new(OnchainFeeEstimator::new());
-
-			let wallet = Arc::new(Wallet::new(
-				bdk_wallet,
-				wallet_persister,
-				Arc::clone(&tx_broadcaster),
-				Arc::clone(&fee_estimator),
-				Arc::clone(&logger),
-			));
-
-			let chain_source = Arc::new(ChainSource::new_esplora(
-				server_url.clone(),
-				Arc::clone(&wallet),
-				Arc::clone(&fee_estimator),
-				Arc::clone(&config),
-				Arc::clone(&logger),
-			));
+	let tx_broadcaster = Arc::new(TransactionBroadcaster::new(Arc::clone(&logger)));
+	let fee_estimator = Arc::new(OnchainFeeEstimator::new());
+	let wallet = Arc::new(Wallet::new(
+		bdk_wallet,
+		wallet_persister,
+		Arc::clone(&tx_broadcaster),
+		Arc::clone(&fee_estimator),
+		Arc::clone(&logger),
+	));
 
-			(wallet, chain_source, tx_broadcaster, fee_estimator)
-		},
+	let chain_source = match chain_data_source_config {
+		Some(ChainDataSourceConfig::Esplora(server_url)) => Arc::new(ChainSource::new_esplora(
+			server_url.clone(),
+			Arc::clone(&wallet),
+			Arc::clone(&fee_estimator),
+			Arc::clone(&tx_broadcaster),
+			Arc::clone(&config),
+			Arc::clone(&logger),
+		)),
 		None => {
 			// Default to Esplora client.
 			let server_url = DEFAULT_ESPLORA_SERVER_URL.to_string();
-			let mut client_builder = esplora_client::Builder::new(&server_url);
-			client_builder = client_builder.timeout(DEFAULT_ESPLORA_CLIENT_TIMEOUT_SECS);
-			let esplora_client = client_builder.build_async().unwrap();
-			let tx_broadcaster =
-				Arc::new(TransactionBroadcaster::new(esplora_client.clone(), Arc::clone(&logger)));
-			let fee_estimator = Arc::new(OnchainFeeEstimator::new());
-
-			let wallet = Arc::new(Wallet::new(
-				bdk_wallet,
-				wallet_persister,
-				Arc::clone(&tx_broadcaster),
-				Arc::clone(&fee_estimator),
-				Arc::clone(&logger),
-			));
-
-			let chain_source = Arc::new(ChainSource::new_esplora(
+			Arc::new(ChainSource::new_esplora(
 				server_url.clone(),
 				Arc::clone(&wallet),
 				Arc::clone(&fee_estimator),
+				Arc::clone(&tx_broadcaster),
 				Arc::clone(&config),
 				Arc::clone(&logger),
-			));
-
-			(wallet, chain_source, tx_broadcaster, fee_estimator)
+			))
 		},
 	};
 

src/chain/mod.rs

@@ -7,17 +7,18 @@
 
 use crate::config::{
 	Config, BDK_CLIENT_CONCURRENCY, BDK_CLIENT_STOP_GAP, BDK_WALLET_SYNC_TIMEOUT_SECS,
-	FEE_RATE_CACHE_UPDATE_TIMEOUT_SECS, LDK_WALLET_SYNC_TIMEOUT_SECS,
+	FEE_RATE_CACHE_UPDATE_TIMEOUT_SECS, LDK_WALLET_SYNC_TIMEOUT_SECS, TX_BROADCAST_TIMEOUT_SECS,
 };
 use crate::fee_estimator::{
 	apply_post_estimation_adjustments, get_all_conf_targets, get_num_block_defaults_for_target,
 	OnchainFeeEstimator,
 };
-use crate::logger::{log_error, log_info, log_trace, FilesystemLogger, Logger};
-use crate::types::Wallet;
+use crate::logger::{log_bytes, log_error, log_info, log_trace, FilesystemLogger, Logger};
+use crate::types::{Broadcaster, Wallet};
 use crate::Error;
 
 use lightning::chain::{Confirm, Filter};
+use lightning::util::ser::Writeable;
 
 use lightning_transaction_sync::EsploraSyncClient;
 
@@ -98,6 +99,7 @@ pub(crate) enum ChainSource {
 		tx_sync: Arc<EsploraSyncClient<Arc<FilesystemLogger>>>,
 		lightning_wallet_sync_status: Mutex<WalletSyncStatus>,
 		fee_estimator: Arc<OnchainFeeEstimator>,
+		tx_broadcaster: Arc<Broadcaster>,
 		config: Arc<Config>,
 		logger: Arc<FilesystemLogger>,
 	},
@@ -106,9 +108,9 @@ pub(crate) enum ChainSource {
 impl ChainSource {
 	pub(crate) fn new_esplora(
 		server_url: String, onchain_wallet: Arc<Wallet>, fee_estimator: Arc<OnchainFeeEstimator>,
-		config: Arc<Config>, logger: Arc<FilesystemLogger>,
+		tx_broadcaster: Arc<Broadcaster>, config: Arc<Config>, logger: Arc<FilesystemLogger>,
 	) -> Self {
-		let mut client_builder = esplora_client::Builder::new(&server_url.clone());
+		let mut client_builder = esplora_client::Builder::new(&server_url);
 		client_builder = client_builder.timeout(DEFAULT_ESPLORA_CLIENT_TIMEOUT_SECS);
 		let esplora_client = client_builder.build_async().unwrap();
 		let tx_sync =
@@ -122,6 +124,7 @@ impl ChainSource {
 			tx_sync,
 			lightning_wallet_sync_status,
 			fee_estimator,
+			tx_broadcaster,
 			config,
 			logger,
 		}
@@ -298,6 +301,82 @@ impl ChainSource {
 			},
 		}
 	}
+
+	pub(crate) async fn process_broadcast_queue(&self) {
+		match self {
+			Self::Esplora { esplora_client, tx_broadcaster, logger, .. } => {
+				let mut receiver = tx_broadcaster.get_broadcast_queue().await;
+				while let Some(next_package) = receiver.recv().await {
+					for tx in &next_package {
+						let txid = tx.compute_txid();
+						let timeout_fut = tokio::time::timeout(
+							Duration::from_secs(TX_BROADCAST_TIMEOUT_SECS),
+							esplora_client.broadcast(tx),
+						);
+						match timeout_fut.await {
+							Ok(res) => match res {
+								Ok(()) => {
+									log_trace!(
+										logger,
+										"Successfully broadcast transaction {}",
+										txid
+									);
+								},
+								Err(e) => match e {
+									esplora_client::Error::Reqwest(err) => {
+										if err.status() == reqwest::StatusCode::from_u16(400).ok() {
+											// Ignore 400, as this just means bitcoind already knows the
+											// transaction.
+											// FIXME: We can further differentiate here based on the error
+											// message which will be available with rust-esplora-client 0.7 and
+											// later.
+										} else {
+											log_error!(
+												logger,
+												"Failed to broadcast due to HTTP connection error: {}",
+												err
+											);
+										}
+										log_trace!(
+											logger,
+											"Failed broadcast transaction bytes: {}",
+											log_bytes!(tx.encode())
+										);
+									},
+									_ => {
+										log_error!(
+											logger,
+											"Failed to broadcast transaction {}: {}",
+											txid,
+											e
+										);
+										log_trace!(
+											logger,
+											"Failed broadcast transaction bytes: {}",
+											log_bytes!(tx.encode())
+										);
+									},
+								},
+							},
+							Err(e) => {
+								log_error!(
+									logger,
+									"Failed to broadcast transaction due to timeout {}: {}",
+									txid,
+									e
+								);
+								log_trace!(
+									logger,
+									"Failed broadcast transaction bytes: {}",
+									log_bytes!(tx.encode())
+								);
+							},
+						}
+					}
+				}
+			},
+		}
+	}
 }
 
 impl Filter for ChainSource {

src/lib.rs

@@ -669,7 +669,7 @@ impl Node {
 		}
 
 		let mut stop_tx_bcast = self.stop_sender.subscribe();
-		let tx_bcaster = Arc::clone(&self.tx_broadcaster);
+		let chain_source = Arc::clone(&self.chain_source);
 		let tx_bcast_logger = Arc::clone(&self.logger);
 		runtime.spawn(async move {
 			// Every second we try to clear our broadcasting queue.
@@ -685,7 +685,7 @@ impl Node {
 							return;
 						}
 						_ = interval.tick() => {
-							tx_bcaster.process_queue().await;
+							chain_source.process_broadcast_queue().await;
 						}
 				}
 			}

src/tx_broadcaster.rs

@@ -5,22 +5,16 @@
 // http://opensource.org/licenses/MIT>, at your option. You may not use this file except in
 // accordance with one or both of these licenses.
 
-use crate::config::TX_BROADCAST_TIMEOUT_SECS;
-use crate::logger::{log_bytes, log_error, log_trace, Logger};
+use crate::logger::{log_error, Logger};
 
 use lightning::chain::chaininterface::BroadcasterInterface;
-use lightning::util::ser::Writeable;
-
-use esplora_client::AsyncClient as EsploraClient;
 
 use bitcoin::Transaction;
 
-use reqwest::StatusCode;
 use tokio::sync::mpsc;
-use tokio::sync::Mutex;
+use tokio::sync::{Mutex, MutexGuard};
 
 use std::ops::Deref;
-use std::time::Duration;
 
 const BCAST_PACKAGE_QUEUE_SIZE: usize = 50;
 
@@ -30,85 +24,20 @@ where
 {
 	queue_sender: mpsc::Sender<Vec<Transaction>>,
 	queue_receiver: Mutex<mpsc::Receiver<Vec<Transaction>>>,
-	esplora_client: EsploraClient,
 	logger: L,
 }
 
 impl<L: Deref> TransactionBroadcaster<L>
 where
 	L::Target: Logger,
 {
-	pub(crate) fn new(esplora_client: EsploraClient, logger: L) -> Self {
+	pub(crate) fn new(logger: L) -> Self {
 		let (queue_sender, queue_receiver) = mpsc::channel(BCAST_PACKAGE_QUEUE_SIZE);
-		Self { queue_sender, queue_receiver: Mutex::new(queue_receiver), esplora_client, logger }
+		Self { queue_sender, queue_receiver: Mutex::new(queue_receiver), logger }
 	}
 
-	pub(crate) async fn process_queue(&self) {
-		let mut receiver = self.queue_receiver.lock().await;
-		while let Some(next_package) = receiver.recv().await {
-			for tx in &next_package {
-				let txid = tx.compute_txid();
-				let timeout_fut = tokio::time::timeout(
-					Duration::from_secs(TX_BROADCAST_TIMEOUT_SECS),
-					self.esplora_client.broadcast(tx),
-				);
-				match timeout_fut.await {
-					Ok(res) => match res {
-						Ok(()) => {
-							log_trace!(self.logger, "Successfully broadcast transaction {}", txid);
-						},
-						Err(e) => match e {
-							esplora_client::Error::Reqwest(err) => {
-								if err.status() == StatusCode::from_u16(400).ok() {
-									// Ignore 400, as this just means bitcoind already knows the
-									// transaction.
-									// FIXME: We can further differentiate here based on the error
-									// message which will be available with rust-esplora-client 0.7 and
-									// later.
-								} else {
-									log_error!(
-										self.logger,
-										"Failed to broadcast due to HTTP connection error: {}",
-										err
-									);
-								}
-								log_trace!(
-									self.logger,
-									"Failed broadcast transaction bytes: {}",
-									log_bytes!(tx.encode())
-								);
-							},
-							_ => {
-								log_error!(
-									self.logger,
-									"Failed to broadcast transaction {}: {}",
-									txid,
-									e
-								);
-								log_trace!(
-									self.logger,
-									"Failed broadcast transaction bytes: {}",
-									log_bytes!(tx.encode())
-								);
-							},
-						},
-					},
-					Err(e) => {
-						log_error!(
-							self.logger,
-							"Failed to broadcast transaction due to timeout {}: {}",
-							txid,
-							e
-						);
-						log_trace!(
-							self.logger,
-							"Failed broadcast transaction bytes: {}",
-							log_bytes!(tx.encode())
-						);
-					},
-				}
-			}
-		}
+	pub(crate) async fn get_broadcast_queue(&self) -> MutexGuard<mpsc::Receiver<Vec<Transaction>>> {
+		self.queue_receiver.lock().await
 	}
 }