Submit packages via bitcoind rpc

Author: tankyleo

src/chain/bitcoind.rs

@@ -596,11 +596,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),
@@ -635,6 +632,40 @@ 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(res) => {
+						// TODO: We'd like to debug assert here the txids, but we sometimes
+						// get 0 txids back
+						log_trace!(self.logger, "Package broadcast result {}", res);
+					},
+					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()));
+					}
+				},
+			}
 		}
 	}
 }
@@ -731,6 +762,32 @@ 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> {
+		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> {
+		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::<serde_json::Value>("submitpackage", &[package_json])
+			.await
+			.map(|value| value.to_string())
+	}
+
 	/// 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(
@@ -1366,6 +1423,45 @@ 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