Submit TRUC packages via all chain sources

tankyleo · tankyleo · commit 195df39a7dff · 2026-07-16T06:46:43.000Z
We rely on the `BroadcasterInterface` contract whereby any
multi-transaction vector must be a single child and its parents.

In a prior commit, we added the guarantee that any packages received
from the broadcast queue are already topologically sorted, and hence
can be passed directly to the `submit_package` Bitcoin Core RPC.

We avoid broadcasting non-TRUC parents-child packages via
`submitpackage` for now to avoid adding a requirement to support
`submitpackage` for users that don't enable zero-fee commitment
channels. We will do so once support for `submitpackage` is more
ubiquitous.
diff --git a/src/chain/bitcoind.rs b/src/chain/bitcoind.rs
@@ -14,6 +14,7 @@ use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
 
 use base64::prelude::BASE64_STANDARD;
 use base64::Engine;
+use bitcoin::transaction::Version;
 use bitcoin::{BlockHash, FeeRate, Network, OutPoint, Transaction, Txid};
 use lightning::chain::chaininterface::ConfirmationTarget as LdkConfirmationTarget;
 use lightning::chain::{BlockLocator, Listen};
@@ -596,7 +597,9 @@ impl BitcoindChainSource {
 		Ok(())
 	}
 
-	fn log_broadcast_error(&self, e: impl core::fmt::Display, txids: &[Txid], txs: &[Transaction]) {
+	fn log_broadcast_error(
+		&self, e: impl core::fmt::Display, txids: &[Txid], txs: &SortedTransactions,
+	) {
 		log_error!(self.logger, "Failed to broadcast transaction(s) {:?}: {}", txids, e);
 		log_trace!(self.logger, "Failed broadcast transaction bytes:");
 		for tx in txs.iter() {
@@ -605,27 +608,48 @@ impl BitcoindChainSource {
 	}
 
 	pub(crate) async fn process_transaction_broadcast(&self, txs: SortedTransactions) {
-		// 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 txs = txs.into_inner();
-		for tx in txs {
-			let txid = tx.compute_txid();
-			let timeout_fut = tokio::time::timeout(
-				Duration::from_secs(DEFAULT_TX_BROADCAST_TIMEOUT_SECS),
-				self.api_client.broadcast_transaction(&tx),
-			);
-			match timeout_fut.await {
-				Ok(res) => match res {
-					Ok(id) => {
-						debug_assert_eq!(id, txid);
-						log_trace!(self.logger, "Successfully broadcast transaction {}", txid);
+		let all_txs_are_v3 = txs.iter().all(|tx| tx.version == Version::non_standard(3));
+		match txs.len() {
+			2.. if all_txs_are_v3 => {
+				let txids: Vec<_> = txs.iter().map(|tx| tx.compute_txid()).collect();
+				let timeout_fut = tokio::time::timeout(
+					Duration::from_secs(DEFAULT_TX_BROADCAST_TIMEOUT_SECS),
+					self.api_client.submit_package(&txs),
+				);
+				match timeout_fut.await {
+					Ok(res) => match res {
+						Ok(result) => {
+							log_trace!(self.logger, "Successfully broadcast package {:?}", txids);
+							log_trace!(self.logger, "Successfully broadcast package {}", result);
+						},
+						Err(e) => self.log_broadcast_error(e, &txids, &txs),
 					},
-					Err(e) => self.log_broadcast_error(e, &[txid], &[tx]),
-				},
-				Err(e) => self.log_broadcast_error(e, &[txid], &[tx]),
-			}
+					Err(e) => self.log_broadcast_error(e, &txids, &txs),
+				}
+			},
+			_ => {
+				for tx in txs.iter() {
+					let txid = tx.compute_txid();
+					let timeout_fut = tokio::time::timeout(
+						Duration::from_secs(DEFAULT_TX_BROADCAST_TIMEOUT_SECS),
+						self.api_client.broadcast_transaction(tx),
+					);
+					match timeout_fut.await {
+						Ok(res) => match res {
+							Ok(id) => {
+								debug_assert_eq!(id, txid);
+								log_trace!(
+									self.logger,
+									"Successfully broadcast transaction {}",
+									txid
+								);
+							},
+							Err(e) => self.log_broadcast_error(e, &[txid], &txs),
+						},
+						Err(e) => self.log_broadcast_error(e, &[txid], &txs),
+					}
+				}
+			},
 		}
 	}
 }
@@ -816,6 +840,38 @@ impl BitcoindClient {
 		rpc_client.call_method::<Txid>("sendrawtransaction", &[tx_json]).await
 	}
 
+	/// Submits the provided package
+	pub(crate) async fn submit_package(
+		&self, package: &SortedTransactions,
+	) -> Result<String, BitcoindClientError> {
+		match self {
+			BitcoindClient::Rpc { rpc_client, .. } => {
+				Self::submit_package_inner(Arc::clone(rpc_client), package)
+					.await
+					.map_err(BitcoindClientError::Rpc)
+			},
+			BitcoindClient::Rest { rpc_client, .. } => {
+				// Bitcoin Core's REST interface does not support submitting packages
+				// so we use the RPC client.
+				Self::submit_package_inner(Arc::clone(rpc_client), package)
+					.await
+					.map_err(BitcoindClientError::Rpc)
+			},
+		}
+	}
+
+	async fn submit_package_inner(
+		rpc_client: Arc<RpcClient>, package: &SortedTransactions,
+	) -> Result<String, RpcClientError> {
+		let package_serialized: Vec<_> =
+			package.iter().map(|tx| bitcoin::consensus::encode::serialize_hex(tx)).collect();
+		let package_json = serde_json::json!(package_serialized);
+		rpc_client
+			.call_method::<SubmitPackageResponse>("submitpackage", &[package_json])
+			.await
+			.map(|response| response.0)
+	}
+
 	/// Retrieve the fee estimate needed for a transaction to begin
 	/// confirmation within the provided `num_blocks`.
 	pub(crate) async fn get_fee_estimate_for_target(
@@ -1367,6 +1423,23 @@ impl TryInto<GetMempoolEntryResponse> for JsonResponse {
 	}
 }
 
+pub struct SubmitPackageResponse(String);
+
+impl TryInto<SubmitPackageResponse> for JsonResponse {
+	type Error = String;
+	fn try_into(self) -> Result<SubmitPackageResponse, String> {
+		let response = self.0.to_string();
+		let res = self.0.as_object().ok_or("Failed to parse submitpackage response".to_string())?;
+
+		match res["package_msg"].as_str() {
+			Some("success") => Ok(SubmitPackageResponse(response)),
+			Some(_) | None => {
+				return Err(response);
+			},
+		}
+	}
+}
+
 #[derive(Debug, Clone)]
 pub(crate) struct MempoolEntry {
 	/// The transaction id
diff --git a/src/chain/electrum.rs b/src/chain/electrum.rs
@@ -16,6 +16,7 @@ use bdk_chain::bdk_core::spk_client::{
 };
 use bdk_electrum::BdkElectrumClient;
 use bdk_wallet::{KeychainKind as BdkKeyChainKind, Update as BdkUpdate};
+use bitcoin::transaction::Version;
 use bitcoin::{FeeRate, Network, Script, ScriptBuf, Transaction, Txid};
 use electrum_client::{
 	Batch, Client as ElectrumClient, ConfigBuilder as ElectrumConfigBuilder, ElectrumApi,
@@ -364,8 +365,14 @@ impl ElectrumChainSource {
 			return;
 		};
 
-		for tx in txs.into_inner() {
-			electrum_client.broadcast(tx).await;
+		let all_txs_are_v3 = txs.iter().all(|tx| tx.version == Version::non_standard(3));
+		match txs.len() {
+			2.. if all_txs_are_v3 => electrum_client.submit_package(txs).await,
+			_ => {
+				for tx in txs.into_inner() {
+					electrum_client.broadcast(tx).await
+				}
+			},
 		}
 	}
 }
@@ -662,6 +669,46 @@ impl ElectrumRuntimeClient {
 		}
 	}
 
+	async fn submit_package(&self, package: SortedTransactions) {
+		let electrum_client = Arc::clone(&self.electrum_client);
+
+		let txids: Vec<_> = package.iter().map(|tx| tx.compute_txid()).collect();
+		let package = Arc::new(package);
+
+		let spawn_fut = self.runtime.spawn_blocking({
+			let package = Arc::clone(&package);
+			move || electrum_client.transaction_broadcast_package(&package)
+		});
+		let timeout_fut = tokio::time::timeout(
+			Duration::from_secs(self.sync_config.timeouts_config.tx_broadcast_timeout_secs),
+			spawn_fut,
+		);
+
+		match timeout_fut.await {
+			Ok(res) => match res {
+				Ok(Ok(result)) => {
+					if result.success {
+						log_trace!(
+							self.logger,
+							"Successfully broadcast transaction(s) {:?}",
+							txids
+						);
+						log_trace!(
+							self.logger,
+							"Successfully broadcast transaction(s) {:?}",
+							result
+						);
+					} else {
+						self.log_broadcast_error(format!("{:?}", result), &txids, &package);
+					}
+				},
+				Ok(Err(e)) => self.log_broadcast_error(e, &txids, &package),
+				Err(e) => self.log_broadcast_error(e, &txids, &package),
+			},
+			Err(e) => self.log_broadcast_error(e, &txids, &package),
+		}
+	}
+
 	async fn get_fee_rate_cache_update(
 		&self,
 	) -> Result<HashMap<ConfirmationTarget, FeeRate>, Error> {
diff --git a/src/chain/esplora.rs b/src/chain/esplora.rs
@@ -11,6 +11,7 @@ use std::sync::{Arc, Mutex};
 use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
 
 use bdk_esplora::EsploraAsyncExt;
+use bitcoin::transaction::Version;
 use bitcoin::{FeeRate, Network, Script, Txid};
 use esplora_client::AsyncClient as EsploraAsyncClient;
 use lightning::chain::{Confirm, Filter, WatchedOutput};
@@ -461,21 +462,61 @@ impl EsploraChainSource {
 	}
 
 	pub(crate) async fn process_transaction_broadcast(&self, txs: SortedTransactions) {
-		for tx in txs.iter() {
-			let txid = tx.compute_txid();
-			let timeout_fut = tokio::time::timeout(
-				Duration::from_secs(self.sync_config.timeouts_config.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);
+		let all_txs_are_v3 = txs.iter().all(|tx| tx.version == Version::non_standard(3));
+		match txs.len() {
+			2.. if all_txs_are_v3 => {
+				let txids: Vec<_> = txs.iter().map(|tx| tx.compute_txid()).collect();
+				let timeout_fut = tokio::time::timeout(
+					Duration::from_secs(self.sync_config.timeouts_config.tx_broadcast_timeout_secs),
+					self.esplora_client.submit_package(&txs, None, None),
+				);
+				match timeout_fut.await {
+					Ok(res) => match res {
+						Ok(result) => {
+							if result.package_msg.eq_ignore_ascii_case("success") {
+								log_trace!(
+									self.logger,
+									"Successfully broadcast transactions {:?}",
+									txids
+								);
+								log_trace!(
+									self.logger,
+									"Successfully broadcast transactions {:?}",
+									result
+								);
+							} else {
+								self.log_broadcast_error(format!("{:?}", result), &txids, &txs);
+							}
+						},
+						Err(e) => self.log_http_error(e, &txids, &txs),
 					},
-					Err(e) => self.log_http_error(e, &[txid], &txs),
-				},
-				Err(e) => self.log_broadcast_error(e, &[txid], &txs),
-			}
+					Err(e) => self.log_broadcast_error(e, &txids, &txs),
+				}
+			},
+			_ => {
+				for tx in txs.iter() {
+					let txid = tx.compute_txid();
+					let timeout_fut = tokio::time::timeout(
+						Duration::from_secs(
+							self.sync_config.timeouts_config.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) => self.log_http_error(e, &[txid], &txs),
+						},
+						Err(e) => self.log_broadcast_error(e, &[txid], &txs),
+					}
+				}
+			},
 		}
 	}
 }
