Introduce TransactionBroadcaster

We introduce a separate `TransactionBroadcaster` which will regularly
process a message queue of transactions to be broadcast. This allows us
to a) decouple broadcasting from the BDK wallet and b) add transactions
to the queue from a blocking context, without the need to schlep around
a separate runtime or mess with it.

Author: tnull

src/builder.rs

@@ -5,6 +5,7 @@ use crate::io::sqlite_store::SqliteStore;
 use crate::logger::{log_error, FilesystemLogger, Logger};
 use crate::payment_store::PaymentStore;
 use crate::peer_store::PeerStore;
+use crate::tx_broadcaster::TransactionBroadcaster;
 use crate::types::{
 	ChainMonitor, ChannelManager, FakeMessageRouter, GossipSync, KeysManager, NetworkGraph,
 	OnionMessenger, PeerManager,
@@ -464,13 +465,17 @@ fn build_with_store_internal<K: KVStore + Sync + Send + 'static>(
 		BuildError::WalletSetupFailed
 	})?;
 
-	let (blockchain, tx_sync) = match chain_data_source_config {
+	let (blockchain, tx_sync, tx_broadcaster) = 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);
-			(blockchain, tx_sync)
+			let tx_broadcaster = Arc::new(TransactionBroadcaster::new(
+				tx_sync.client().clone(),
+				Arc::clone(&logger),
+			));
+			(blockchain, tx_sync, tx_broadcaster)
 		}
 		None => {
 			// Default to Esplora client.
@@ -479,18 +484,26 @@ fn build_with_store_internal<K: KVStore + Sync + Send + 'static>(
 			let blockchain =
 				EsploraBlockchain::from_client(tx_sync.client().clone(), BDK_CLIENT_STOP_GAP)
 					.with_concurrency(BDK_CLIENT_CONCURRENCY);
-			(blockchain, tx_sync)
+			let tx_broadcaster = Arc::new(TransactionBroadcaster::new(
+				tx_sync.client().clone(),
+				Arc::clone(&logger),
+			));
+			(blockchain, tx_sync, tx_broadcaster)
 		}
 	};
 
 	let runtime = Arc::new(RwLock::new(None));
-	let wallet =
-		Arc::new(Wallet::new(blockchain, bdk_wallet, Arc::clone(&runtime), Arc::clone(&logger)));
+	let wallet = Arc::new(Wallet::new(
+		blockchain,
+		bdk_wallet,
+		Arc::clone(&tx_broadcaster),
+		Arc::clone(&logger),
+	));
 
 	// Initialize the ChainMonitor
 	let chain_monitor: Arc<ChainMonitor<K>> = Arc::new(chainmonitor::ChainMonitor::new(
 		Some(Arc::clone(&tx_sync)),
-		Arc::clone(&wallet),
+		Arc::clone(&tx_broadcaster),
 		Arc::clone(&logger),
 		Arc::clone(&wallet),
 		Arc::clone(&kv_store),
@@ -594,7 +607,7 @@ fn build_with_store_internal<K: KVStore + Sync + Send + 'static>(
 				Arc::clone(&keys_manager),
 				Arc::clone(&wallet),
 				Arc::clone(&chain_monitor),
-				Arc::clone(&wallet),
+				Arc::clone(&tx_broadcaster),
 				Arc::clone(&router),
 				Arc::clone(&logger),
 				user_config,
@@ -618,7 +631,7 @@ fn build_with_store_internal<K: KVStore + Sync + Send + 'static>(
 			channelmanager::ChannelManager::new(
 				Arc::clone(&wallet),
 				Arc::clone(&chain_monitor),
-				Arc::clone(&wallet),
+				Arc::clone(&tx_broadcaster),
 				Arc::clone(&router),
 				Arc::clone(&logger),
 				Arc::clone(&keys_manager),
@@ -767,6 +780,7 @@ fn build_with_store_internal<K: KVStore + Sync + Send + 'static>(
 		config,
 		wallet,
 		tx_sync,
+		tx_broadcaster,
 		event_queue,
 		channel_manager,
 		chain_monitor,

src/event.rs

@@ -1,4 +1,4 @@
-use crate::types::Wallet;
+use crate::types::{Broadcaster, Wallet};
 use crate::{hex_utils, ChannelManager, Config, Error, KeysManager, NetworkGraph, UserChannelId};
 
 use crate::payment_store::{
@@ -241,9 +241,10 @@ pub(crate) struct EventHandler<K: KVStore + Sync + Send, L: Deref>
 where
 	L::Target: Logger,
 {
-	wallet: Arc<Wallet>,
 	event_queue: Arc<EventQueue<K, L>>,
+	wallet: Arc<Wallet>,
 	channel_manager: Arc<ChannelManager<K>>,
+	tx_broadcaster: Arc<Broadcaster>,
 	network_graph: Arc<NetworkGraph>,
 	keys_manager: Arc<KeysManager>,
 	payment_store: Arc<PaymentStore<K, L>>,
@@ -257,15 +258,17 @@ where
 	L::Target: Logger,
 {
 	pub fn new(
-		wallet: Arc<Wallet>, event_queue: Arc<EventQueue<K, L>>,
-		channel_manager: Arc<ChannelManager<K>>, network_graph: Arc<NetworkGraph>,
-		keys_manager: Arc<KeysManager>, payment_store: Arc<PaymentStore<K, L>>,
+		event_queue: Arc<EventQueue<K, L>>, wallet: Arc<Wallet>,
+		channel_manager: Arc<ChannelManager<K>>, tx_broadcaster: Arc<Broadcaster>,
+		network_graph: Arc<NetworkGraph>, keys_manager: Arc<KeysManager>,
+		payment_store: Arc<PaymentStore<K, L>>,
 		runtime: Arc<RwLock<Option<tokio::runtime::Runtime>>>, logger: L, config: Arc<Config>,
 	) -> Self {
 		Self {
 			event_queue,
 			wallet,
 			channel_manager,
+			tx_broadcaster,
 			network_graph,
 			keys_manager,
 			payment_store,
@@ -598,7 +601,9 @@ where
 				);
 
 				match res {
-					Ok(Some(spending_tx)) => self.wallet.broadcast_transactions(&[&spending_tx]),
+					Ok(Some(spending_tx)) => {
+						self.tx_broadcaster.broadcast_transactions(&[&spending_tx])
+					}
 					Ok(None) => {
 						log_debug!(self.logger, "Omitted spending static outputs: {:?}", outputs);
 					}

src/lib.rs

@@ -86,6 +86,7 @@ mod payment_store;
 mod peer_store;
 #[cfg(test)]
 mod test;
+mod tx_broadcaster;
 mod types;
 #[cfg(feature = "uniffi")]
 mod uniffi_types;
@@ -119,7 +120,8 @@ use payment_store::PaymentStore;
 pub use payment_store::{PaymentDetails, PaymentDirection, PaymentStatus};
 use peer_store::{PeerInfo, PeerStore};
 use types::{
-	ChainMonitor, ChannelManager, KeysManager, NetworkGraph, PeerManager, Router, Scorer, Wallet,
+	Broadcaster, ChainMonitor, ChannelManager, KeysManager, NetworkGraph, PeerManager, Router,
+	Scorer, Wallet,
 };
 pub use types::{ChannelDetails, PeerDetails, UserChannelId};
 
@@ -287,6 +289,7 @@ pub struct Node<K: KVStore + Sync + Send + 'static> {
 	config: Arc<Config>,
 	wallet: Arc<Wallet>,
 	tx_sync: Arc<EsploraSyncClient<Arc<FilesystemLogger>>>,
+	tx_broadcaster: Arc<Broadcaster>,
 	event_queue: Arc<EventQueue<K, Arc<FilesystemLogger>>>,
 	channel_manager: Arc<ChannelManager<K>>,
 	chain_monitor: Arc<ChainMonitor<K>>,
@@ -647,10 +650,29 @@ impl<K: KVStore + Sync + Send + 'static> Node<K> {
 			}
 		});
 
+		let mut stop_tx_bcast = self.stop_receiver.clone();
+		let tx_bcaster = Arc::clone(&self.tx_broadcaster);
+		runtime.spawn(async move {
+			// Every second we try to clear our broadcasting queue.
+			let mut interval = tokio::time::interval(Duration::from_secs(1));
+			interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
+			loop {
+				tokio::select! {
+						_ = stop_tx_bcast.changed() => {
+							return;
+						}
+						_ = interval.tick() => {
+							tx_bcaster.process_next_package().await;
+						}
+				}
+			}
+		});
+
 		let event_handler = Arc::new(EventHandler::new(
-			Arc::clone(&self.wallet),
 			Arc::clone(&self.event_queue),
+			Arc::clone(&self.wallet),
 			Arc::clone(&self.channel_manager),
+			Arc::clone(&self.tx_broadcaster),
 			Arc::clone(&self.network_graph),
 			Arc::clone(&self.keys_manager),
 			Arc::clone(&self.payment_store),

src/tx_broadcaster.rs

@@ -0,0 +1,67 @@
+use crate::logger::{log_error, log_trace, Logger};
+
+use lightning::chain::chaininterface::BroadcasterInterface;
+
+use esplora_client::AsyncClient as EsploraClient;
+
+use bitcoin::Transaction;
+
+use tokio::sync::mpsc;
+use tokio::sync::Mutex;
+
+use std::ops::Deref;
+
+const BCAST_PACKAGE_QUEUE_SIZE: usize = 50;
+
+pub(crate) struct TransactionBroadcaster<L: Deref>
+where
+	L::Target: Logger,
+{
+	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 {
+		let (queue_sender, queue_receiver) = mpsc::channel(BCAST_PACKAGE_QUEUE_SIZE);
+		Self { queue_sender, queue_receiver: Mutex::new(queue_receiver), esplora_client, logger }
+	}
+
+	pub(crate) async fn process_next_package(&self) {
+		let mut receiver = self.queue_receiver.lock().await;
+		if let Some(next_package) = receiver.recv().await {
+			for tx in &next_package {
+				match self.esplora_client.broadcast(tx).await {
+					Ok(()) => {
+						log_trace!(self.logger, "Successfully broadcast transaction {}", tx.txid());
+					}
+					Err(e) => {
+						log_error!(
+							self.logger,
+							"Failed to broadcast transaction {}: {}",
+							tx.txid(),
+							e
+						);
+					}
+				}
+			}
+		}
+	}
+}
+
+impl<L: Deref> BroadcasterInterface for TransactionBroadcaster<L>
+where
+	L::Target: Logger,
+{
+	fn broadcast_transactions(&self, txs: &[&Transaction]) {
+		let package = txs.iter().map(|&t| t.clone()).collect::<Vec<Transaction>>();
+		self.queue_sender.try_send(package).unwrap_or_else(|e| {
+			log_error!(self.logger, "Failed to broadcast transactions: {}", e);
+		});
+	}
+}

src/types.rs

@@ -24,7 +24,7 @@ use std::sync::{Arc, Mutex, RwLock};
 pub(crate) type ChainMonitor<K> = chainmonitor::ChainMonitor<
 	InMemorySigner,
 	Arc<EsploraSyncClient<Arc<FilesystemLogger>>>,
-	Arc<Wallet>,
+	Arc<Broadcaster>,
 	Arc<Wallet>,
 	Arc<FilesystemLogger>,
 	Arc<K>,
@@ -42,7 +42,7 @@ pub(crate) type PeerManager<K> = lightning::ln::peer_handler::PeerManager<
 
 pub(crate) type ChannelManager<K> = lightning::ln::channelmanager::ChannelManager<
 	Arc<ChainMonitor<K>>,
-	Arc<Wallet>,
+	Arc<Broadcaster>,
 	Arc<KeysManager>,
 	Arc<KeysManager>,
 	Arc<KeysManager>,
@@ -51,11 +51,16 @@ pub(crate) type ChannelManager<K> = lightning::ln::channelmanager::ChannelManage
 	Arc<FilesystemLogger>,
 >;
 
+pub(crate) type Broadcaster = crate::tx_broadcaster::TransactionBroadcaster<Arc<FilesystemLogger>>;
+
 pub(crate) type Wallet =
-	crate::wallet::Wallet<bdk::database::SqliteDatabase, Arc<FilesystemLogger>>;
+	crate::wallet::Wallet<bdk::database::SqliteDatabase, Arc<Broadcaster>, Arc<FilesystemLogger>>;
 
-pub(crate) type KeysManager =
-	crate::wallet::WalletKeysManager<bdk::database::SqliteDatabase, Arc<FilesystemLogger>>;
+pub(crate) type KeysManager = crate::wallet::WalletKeysManager<
+	bdk::database::SqliteDatabase,
+	Arc<Broadcaster>,
+	Arc<FilesystemLogger>,
+>;
 
 pub(crate) type Router = DefaultRouter<
 	Arc<NetworkGraph>,