Submit packages via bitcoind rpc

Author: tankyleo

src/chain/bitcoind.rs

@@ -582,11 +582,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),
@@ -621,6 +618,48 @@ 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((package_msg, mut ids)) => {
+						// TODO: seems we don't get the txids back in the same order...
+						ids.sort_unstable();
+						let mut sorted_txids = txids.clone();
+						sorted_txids.sort_unstable();
+						debug_assert_eq!(ids, sorted_txids);
+						log_trace!(
+							self.logger,
+							"Package broadcast message {}, txids: {:?}",
+							package_msg,
+							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()));
+					}
+				},
+			}
 		}
 	}
 }
@@ -717,6 +756,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<(String, 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<(String, 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.package_msg, resp.txids))
+	}
+
 	/// 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(
@@ -1352,6 +1419,43 @@ impl TryInto<GetMempoolEntryResponse> for JsonResponse {
 	}
 }
 
+pub struct SubmitPackageResponse {
+	package_msg: String,
+	txids: Vec<Txid>,
+}
+
+impl TryInto<SubmitPackageResponse> for JsonResponse {
+	type Error = std::io::Error;
+	fn try_into(self) -> std::io::Result<SubmitPackageResponse> {
+		let package_msg = self.0["package_msg"]
+			.as_str()
+			.ok_or(std::io::Error::new(
+				std::io::ErrorKind::Other,
+				"Failed to parse submitpackage response",
+			))?
+			.to_string();
+		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 { package_msg, txids })
+	}
+}
+
 #[derive(Debug, Clone)]
 pub(crate) struct MempoolEntry {
 	/// The transaction id