Submit packages via bitcoind rpc

Author: tankyleo

src/chain/bitcoind.rs

@@ -532,11 +532,8 @@ impl BitcoindChainSource {
 	}
 
 	pub(crate) async fn process_broadcast_package(&self, package: Vec<Transaction>) {
-		// 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.
-		for tx in &package {
+		if package.len() == 1 {
+			let tx = &package[0];
 			let txid = tx.compute_txid();
 			let timeout_fut = tokio::time::timeout(
 				Duration::from_secs(TX_BROADCAST_TIMEOUT_SECS),
@@ -571,6 +568,39 @@ impl BitcoindChainSource {
 					);
 				},
 			}
+		} else if package.len() > 1 {
+			let txids: Vec<_> = package.iter().map(|tx| tx.compute_txid()).collect();
+			let timeout_fut = tokio::time::timeout(
+				Duration::from_secs(TX_BROADCAST_TIMEOUT_SECS),
+				self.api_client.submit_package(&package),
+			);
+			match timeout_fut.await {
+				Ok(res) => match res {
+					Ok(ids) => {
+						debug_assert_eq!(ids, txids);
+						log_trace!(self.logger, "Successfully broadcast package {:?}", txids);
+					},
+					Err(e) => {
+						log_error!(self.logger, "Failed to broadcast package {:?}: {}", txids, e);
+						log_trace!(self.logger, "Failed broadcast package bytes:");
+						for tx in package {
+							log_trace!(self.logger, "{}", log_bytes!(tx.encode()));
+						}
+					},
+				},
+				Err(e) => {
+					log_error!(
+						self.logger,
+						"Failed to broadcast package due to timeout {:?}: {}",
+						txids,
+						e
+					);
+					log_trace!(self.logger, "Failed broadcast package bytes:");
+					for tx in package {
+						log_trace!(self.logger, "{}", log_bytes!(tx.encode()));
+					}
+				},
+			}
 		}
 	}
 }
@@ -667,6 +697,34 @@ impl BitcoindClient {
 		rpc_client.call_method::<Txid>("sendrawtransaction", &[tx_json]).await
 	}
 
+	/// Submits the provided package
+	pub(crate) async fn submit_package(
+		&self, package: &[Transaction],
+	) -> std::io::Result<Vec<Txid>> {
+		match self {
+			BitcoindClient::Rpc { rpc_client, .. } => {
+				Self::submit_package_inner(Arc::clone(rpc_client), package).await
+			},
+			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
+			},
+		}
+	}
+
+	async fn submit_package_inner(
+		rpc_client: Arc<RpcClient>, package: &[Transaction],
+	) -> std::io::Result<Vec<Txid>> {
+		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(|resp| resp.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(
@@ -1302,6 +1360,33 @@ impl TryInto<GetMempoolEntryResponse> for JsonResponse {
 	}
 }
 
+pub struct SubmitPackageResponse(Vec<Txid>);
+
+impl TryInto<SubmitPackageResponse> for JsonResponse {
+	type Error = std::io::Error;
+	fn try_into(self) -> std::io::Result<SubmitPackageResponse> {
+		let tx_results = self.0["tx-results"].as_object().ok_or(std::io::Error::new(
+			std::io::ErrorKind::Other,
+			"Failed to parse submitpackage response",
+		))?;
+		let mut txids = Vec::with_capacity(tx_results.len());
+		for tx_result in tx_results.values() {
+			let txid_string = tx_result["txid"].as_str().ok_or(std::io::Error::new(
+				std::io::ErrorKind::Other,
+				"Failed to parse submitpackage response",
+			))?;
+			let txid: Txid = txid_string.parse().map_err(|_| {
+				std::io::Error::new(
+					std::io::ErrorKind::Other,
+					"Failed to parse submitpackage response",
+				)
+			})?;
+			txids.push(txid);
+		}
+		Ok(SubmitPackageResponse(txids))
+	}
+}
+
 #[derive(Debug, Clone)]
 pub(crate) struct MempoolEntry {
 	/// The transaction id