Implement broadcast queue processing

Author: tnull

Cargo.toml

@@ -73,6 +73,7 @@ tokio = { version = "1.37", default-features = false, features = [ "rt-multi-thr
 esplora-client = { version = "0.9", default-features = false }
 libc = "0.2"
 uniffi = { version = "0.26.0", features = ["build"], optional = true }
+serde_json = { version = "1.0.128", default-features = false, features = ["std"] }
 
 [target.'cfg(vss)'.dependencies]
 vss-client = "0.3"

src/chain/bitcoind_rpc.rs

@@ -16,7 +16,7 @@ use lightning_block_sync::{
 	AsyncBlockSourceResult, BlockData, BlockHeaderData, BlockSource, Cache,
 };
 
-use bitcoin::BlockHash;
+use bitcoin::{BlockHash, Transaction, Txid};
 
 use base64::prelude::{Engine, BASE64_STANDARD};
 
@@ -40,6 +40,12 @@ impl BitcoindRpcClient {
 
 		Self { rpc_client }
 	}
+
+	pub(crate) async fn broadcast_transaction(&self, tx: &Transaction) -> std::io::Result<Txid> {
+		let tx_serialized = bitcoin::consensus::encode::serialize_hex(tx);
+		let tx_json = serde_json::json!(tx_serialized);
+		self.rpc_client.call_method::<Txid>("sendrawtransaction", &vec![tx_json]).await
+	}
 }
 
 impl BlockSource for BitcoindRpcClient {

src/chain/mod.rs

@@ -781,7 +781,60 @@ impl ChainSource {
 					}
 				}
 			},
-			Self::BitcoindRpc { .. } => todo!(),
+			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
+				// `rust-bitcoind-json-rpc` crate rather than just broadcasting individual
+				// transactions.
+				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),
+							bitcoind_rpc_client.broadcast_transaction(tx),
+						);
+						match timeout_fut.await {
+							Ok(res) => match res {
+								Ok(id) => {
+									debug_assert_eq!(id, txid);
+									log_trace!(
+										logger,
+										"Successfully broadcast transaction {}",
+										txid
+									);
+								},
+								Err(e) => {
+									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())
+								);
+							},
+						}
+					}
+				}
+			},
 		}
 	}
 }