Implement process_broadcast_queue for ChainSource::Electrum

Author: tnull

src/chain/electrum.rs

@@ -5,19 +5,22 @@
 // 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::error::Error;
-use crate::logger::{log_error, LdkLogger, Logger};
+use crate::logger::{log_bytes, log_error, log_trace, LdkLogger, Logger};
 
 use lightning::chain::{Filter, WatchedOutput};
+use lightning::util::ser::Writeable;
 use lightning_transaction_sync::ElectrumSyncClient;
 
 use bdk_electrum::BdkElectrumClient;
 
-use electrum_client::Client as ElectrumClient;
+use electrum_client::{Client as ElectrumClient, ElectrumApi};
 
-use bitcoin::{Script, Txid};
+use bitcoin::{Script, Transaction, Txid};
 
 use std::sync::Arc;
+use std::time::Duration;
 
 pub(crate) struct ElectrumRuntimeClient {
 	electrum_client: Arc<ElectrumClient>,
@@ -48,6 +51,48 @@ impl ElectrumRuntimeClient {
 		);
 		Ok(Self { electrum_client, bdk_electrum_client, tx_sync, runtime, logger })
 	}
+
+	pub(crate) async fn broadcast(&self, tx: Transaction) {
+		let electrum_client = Arc::clone(&self.electrum_client);
+
+		let txid = tx.compute_txid();
+		let tx_bytes = tx.encode();
+
+		let spawn_fut =
+			self.runtime.spawn_blocking(move || electrum_client.transaction_broadcast(&tx));
+
+		let timeout_fut =
+			tokio::time::timeout(Duration::from_secs(TX_BROADCAST_TIMEOUT_SECS), spawn_fut);
+
+		match timeout_fut.await {
+			Ok(res) => match res {
+				Ok(_) => {
+					log_trace!(self.logger, "Successfully broadcast transaction {}", txid);
+				},
+				Err(e) => {
+					log_error!(self.logger, "Failed to broadcast transaction {}: {}", txid, e);
+					log_trace!(
+						self.logger,
+						"Failed broadcast transaction bytes: {}",
+						log_bytes!(tx_bytes)
+					);
+				},
+			},
+			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_bytes)
+				);
+			},
+		}
+	}
 }
 
 impl Filter for ElectrumRuntimeClient {

src/chain/mod.rs

@@ -127,7 +127,7 @@ pub(crate) enum ChainSource {
 	Electrum {
 		server_url: String,
 		sync_config: ElectrumSyncConfig,
-		electrum_runtime_client: RwLock<Option<ElectrumRuntimeClient>>,
+		electrum_runtime_client: RwLock<Option<Arc<ElectrumRuntimeClient>>>,
 		registered_txs_cache: Mutex<Vec<(Txid, ScriptBuf)>>,
 		registered_outputs_cache: Mutex<Vec<WatchedOutput>>,
 		onchain_wallet: Arc<Wallet>,
@@ -271,7 +271,7 @@ impl ChainSource {
 					}
 				}
 
-				*locked_client = Some(client);
+				*locked_client = Some(Arc::new(client));
 			},
 			_ => {
 				// Nothing to do for other chain sources.
@@ -1190,7 +1190,25 @@ impl ChainSource {
 					}
 				}
 			},
-			Self::Electrum { .. } => todo!(),
+			Self::Electrum { electrum_runtime_client, tx_broadcaster, .. } => {
+				let electrum_client: Arc<ElectrumRuntimeClient> =
+					if let Some(client) = electrum_runtime_client.read().unwrap().as_ref() {
+						Arc::clone(client)
+					} else {
+						debug_assert!(
+							false,
+							"We should have started the chain source before broadcasting"
+						);
+						return;
+					};
+
+				let mut receiver = tx_broadcaster.get_broadcast_queue().await;
+				while let Some(next_package) = receiver.recv().await {
+					for tx in next_package {
+						electrum_client.broadcast(tx).await;
+					}
+				}
+			},
 			Self::BitcoindRpc { bitcoind_rpc_client, tx_broadcaster, logger, .. } => {
 				// While it's a bit unclear when we'd be able to lean on Bitcoin Core >v28
 				// features, we should eventually switch to use `submitpackage` via the