Sort packages received via BroadcasterInterface

Implementations of `BroadcasterInterface` cannot assume any topological
ordering on the transactions received, so here we order the received
transactions before adding them to the broadcast queue. Any consumers of
the queue can now assume all transactions received to be topologically
sorted.

Codex wrote the tests.

Author: tankyleo

src/chain/bitcoind.rs

@@ -41,6 +41,7 @@ use crate::fee_estimator::{
 };
 use crate::io::utils::update_and_persist_node_metrics;
 use crate::logger::{log_bytes, log_debug, log_error, log_info, log_trace, LdkLogger, Logger};
+use crate::tx_broadcaster::SortedTransactions;
 use crate::types::{ChainMonitor, ChannelManager, DynStore, Sweeper, Wallet};
 use crate::{Error, PersistedNodeMetrics};
 
@@ -595,12 +596,12 @@ impl BitcoindChainSource {
 		Ok(())
 	}
 
-	pub(crate) async fn process_broadcast_package(&self, package: Vec<Transaction>) {
+	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.
-		for tx in &package {
+		for tx in txs.iter() {
 			let txid = tx.compute_txid();
 			let timeout_fut = tokio::time::timeout(
 				Duration::from_secs(DEFAULT_TX_BROADCAST_TIMEOUT_SECS),

src/chain/electrum.rs

@@ -37,6 +37,7 @@ use crate::fee_estimator::{
 use crate::io::utils::update_and_persist_node_metrics;
 use crate::logger::{log_bytes, log_debug, log_error, log_trace, log_warn, LdkLogger, Logger};
 use crate::runtime::Runtime;
+use crate::tx_broadcaster::SortedTransactions;
 use crate::types::{ChainMonitor, ChannelManager, DynStore, Sweeper, Wallet};
 use crate::PersistedNodeMetrics;
 
@@ -353,7 +354,7 @@ impl ElectrumChainSource {
 		}
 	}
 
-	pub(crate) async fn process_broadcast_package(&self, package: Vec<Transaction>) {
+	pub(crate) async fn process_transaction_broadcast(&self, txs: SortedTransactions) {
 		let electrum_client: Arc<ElectrumRuntimeClient> = if let Some(client) =
 			self.electrum_runtime_status.read().expect("lock").client().as_ref()
 		{
@@ -363,7 +364,7 @@ impl ElectrumChainSource {
 			return;
 		};
 
-		for tx in package {
+		for tx in txs.into_inner() {
 			electrum_client.broadcast(tx).await;
 		}
 	}

src/chain/esplora.rs

@@ -11,7 +11,7 @@ use std::sync::{Arc, Mutex};
 use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
 
 use bdk_esplora::EsploraAsyncExt;
-use bitcoin::{FeeRate, Network, Script, Transaction, Txid};
+use bitcoin::{FeeRate, Network, Script, Txid};
 use esplora_client::AsyncClient as EsploraAsyncClient;
 use lightning::chain::{Confirm, Filter, WatchedOutput};
 use lightning::util::ser::Writeable;
@@ -28,6 +28,7 @@ use crate::fee_estimator::{
 };
 use crate::io::utils::update_and_persist_node_metrics;
 use crate::logger::{log_bytes, log_debug, log_error, log_trace, log_warn, LdkLogger, Logger};
+use crate::tx_broadcaster::SortedTransactions;
 use crate::types::{ChainMonitor, ChannelManager, DynStore, Sweeper, Wallet};
 use crate::{Error, PersistedNodeMetrics};
 
@@ -410,8 +411,8 @@ impl EsploraChainSource {
 		Ok(())
 	}
 
-	pub(crate) async fn process_broadcast_package(&self, package: Vec<Transaction>) {
-		for tx in &package {
+	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),

src/chain/mod.rs

@@ -13,7 +13,7 @@ use std::collections::{HashMap, HashSet};
 use std::sync::{Arc, Mutex};
 use std::time::Duration;
 
-use bitcoin::{Script, Transaction, Txid};
+use bitcoin::{Script, Txid};
 use lightning::chain::{BlockLocator, Filter};
 
 use crate::chain::bitcoind::{BitcoindChainSource, UtxoSourceClient};
@@ -518,16 +518,16 @@ impl ChainSource {
 							continue;
 						},
 					};
-					let txs: Vec<Transaction> = package.into_transactions();
+					let package = package.into_sorted_transactions();
 					match &self.kind {
 						ChainSourceKind::Esplora(esplora_chain_source) => {
-							esplora_chain_source.process_broadcast_package(txs).await
+							esplora_chain_source.process_transaction_broadcast(package).await
 						},
 						ChainSourceKind::Electrum(electrum_chain_source) => {
-							electrum_chain_source.process_broadcast_package(txs).await
+							electrum_chain_source.process_transaction_broadcast(package).await
 						},
 						ChainSourceKind::Bitcoind(bitcoind_chain_source) => {
-							bitcoind_chain_source.process_broadcast_package(txs).await
+							bitcoind_chain_source.process_transaction_broadcast(package).await
 						},
 					}
 				}

src/tx_broadcaster.rs

@@ -43,8 +43,52 @@ impl BroadcastPackage {
 	}
 
 	/// Consumes the package into its transactions, ready for the chain client.
-	pub(crate) fn into_transactions(self) -> Vec<Transaction> {
-		self.0.into_iter().map(|(tx, _)| tx).collect()
+	pub(crate) fn into_sorted_transactions(self) -> SortedTransactions {
+		let txs = self.0.into_iter().map(|(tx, _)| tx).collect();
+		SortedTransactions::sort_parents_child_package_topologically(txs)
+	}
+}
+
+pub(crate) struct SortedTransactions(Vec<Transaction>);
+
+impl SortedTransactions {
+	pub(crate) fn sort_parents_child_package_topologically(
+		mut txs: Vec<Transaction>,
+	) -> SortedTransactions {
+		if txs.len() == 0 || txs.len() == 1 {
+			return SortedTransactions(txs);
+		}
+		let txids: Vec<_> = txs.iter().map(|tx| tx.compute_txid()).collect();
+		let any_spends_from_package = |tx: &Transaction| -> bool {
+			tx.input.iter().any(|input| txids.contains(&input.previous_output.txid))
+		};
+		txs.sort_by_key(any_spends_from_package);
+
+		#[cfg(debug_assertions)]
+		{
+			let child = txs.last().expect("txs is not empty");
+			let child_input_txids: Vec<_> =
+				child.input.iter().map(|input| input.previous_output.txid).collect();
+			let parents = &txs[..txs.len() - 1];
+			let parent_txids: Vec<_> = parents.iter().map(|parent| parent.compute_txid()).collect();
+			// Make sure all the parent txids are parents of the child transaction
+			debug_assert!(parent_txids.iter().all(|txid| child_input_txids.contains(&txid)));
+			// Make sure there are no grandparents
+			debug_assert_eq!(txs.iter().filter(|tx| any_spends_from_package(tx)).count(), 1);
+		}
+
+		SortedTransactions(txs)
+	}
+
+	pub(crate) fn into_inner(self) -> Vec<Transaction> {
+		self.0
+	}
+}
+
+impl Deref for SortedTransactions {
+	type Target = Vec<Transaction>;
+	fn deref(&self) -> &Self::Target {
+		&self.0
 	}
 }
 
@@ -123,3 +167,151 @@ where
 		});
 	}
 }
+
+#[cfg(test)]
+mod tests {
+	use bitcoin::hashes::Hash;
+	use bitcoin::{Amount, OutPoint, ScriptBuf, Sequence, Transaction, TxIn, TxOut, Txid, Witness};
+
+	use super::SortedTransactions;
+
+	fn txin(txid: Txid, vout: u32) -> TxIn {
+		TxIn {
+			previous_output: OutPoint { txid, vout },
+			script_sig: ScriptBuf::new(),
+			sequence: Sequence::MAX,
+			witness: Witness::new(),
+		}
+	}
+
+	fn txout(value_sat: u64) -> TxOut {
+		TxOut { value: Amount::from_sat(value_sat), script_pubkey: ScriptBuf::new() }
+	}
+
+	fn parent_tx(seed: u8) -> Transaction {
+		Transaction {
+			version: bitcoin::transaction::Version::TWO,
+			lock_time: bitcoin::absolute::LockTime::ZERO,
+			input: vec![txin(Txid::from_byte_array([seed; 32]), 0)],
+			output: vec![txout(1_000 + u64::from(seed))],
+		}
+	}
+
+	fn child_tx(parents: &[&Transaction]) -> Transaction {
+		Transaction {
+			version: bitcoin::transaction::Version::TWO,
+			lock_time: bitcoin::absolute::LockTime::ZERO,
+			input: parents
+				.iter()
+				.enumerate()
+				.map(|(idx, parent)| txin(parent.compute_txid(), idx as u32))
+				.collect(),
+			output: vec![txout(1_000)],
+		}
+	}
+
+	fn assert_parents_before_child(
+		txs: &[Transaction], expected_child: Txid, expected_parents: &[Txid],
+	) {
+		assert_eq!(txs.last().map(Transaction::compute_txid), Some(expected_child));
+		assert_eq!(txs.len(), expected_parents.len() + 1);
+
+		let parent_txids =
+			txs[..txs.len() - 1].iter().map(Transaction::compute_txid).collect::<Vec<_>>();
+		for expected_parent in expected_parents {
+			assert!(parent_txids.contains(expected_parent));
+		}
+	}
+
+	#[test]
+	fn topological_sort_leaves_sorted_package_unchanged() {
+		let parent_a = parent_tx(1);
+		let parent_b = parent_tx(2);
+		let child = child_tx(&[&parent_a, &parent_b]);
+
+		let original_txids =
+			[parent_a.compute_txid(), parent_b.compute_txid(), child.compute_txid()];
+		let txs = vec![parent_a, parent_b, child];
+
+		let package = SortedTransactions::sort_parents_child_package_topologically(txs);
+
+		assert_eq!(
+			package.iter().map(Transaction::compute_txid).collect::<Vec<_>>(),
+			original_txids
+		);
+	}
+
+	#[test]
+	fn topological_sort_moves_single_parent_child_from_front_to_end() {
+		let parent = parent_tx(1);
+		let child = child_tx(&[&parent]);
+		let parent_txids = [parent.compute_txid()];
+		let child_txid = child.compute_txid();
+		let txs = vec![child, parent];
+
+		let package = SortedTransactions::sort_parents_child_package_topologically(txs);
+
+		assert_parents_before_child(&package, child_txid, &parent_txids);
+	}
+
+	#[test]
+	fn topological_sort_moves_child_from_front_to_end() {
+		let parent_a = parent_tx(1);
+		let parent_b = parent_tx(2);
+		let child = child_tx(&[&parent_a, &parent_b]);
+		let parent_txids = [parent_a.compute_txid(), parent_b.compute_txid()];
+		let child_txid = child.compute_txid();
+		let txs = vec![child, parent_a, parent_b];
+
+		let package = SortedTransactions::sort_parents_child_package_topologically(txs);
+
+		assert_parents_before_child(&package, child_txid, &parent_txids);
+	}
+
+	#[test]
+	fn topological_sort_moves_child_from_front_with_multiple_parents_to_end() {
+		let parent_a = parent_tx(1);
+		let parent_b = parent_tx(2);
+		let parent_c = parent_tx(3);
+		let child = child_tx(&[&parent_a, &parent_b, &parent_c]);
+		let parent_txids =
+			[parent_a.compute_txid(), parent_b.compute_txid(), parent_c.compute_txid()];
+		let child_txid = child.compute_txid();
+		let txs = vec![child, parent_a, parent_b, parent_c];
+
+		let package = SortedTransactions::sort_parents_child_package_topologically(txs);
+
+		assert_parents_before_child(&package, child_txid, &parent_txids);
+	}
+
+	#[test]
+	fn topological_sort_moves_child_from_middle_to_end() {
+		let parent_a = parent_tx(1);
+		let parent_b = parent_tx(2);
+		let child = child_tx(&[&parent_a, &parent_b]);
+		let parent_txids = [parent_a.compute_txid(), parent_b.compute_txid()];
+		let child_txid = child.compute_txid();
+		let txs = vec![parent_a, child, parent_b];
+
+		let package = SortedTransactions::sort_parents_child_package_topologically(txs);
+
+		assert_parents_before_child(&package, child_txid, &parent_txids);
+	}
+
+	#[test]
+	fn topological_sort_leaves_single_transaction_package_unchanged() {
+		let parent = parent_tx(1);
+		let parent_txid = parent.compute_txid();
+		let txs = vec![parent];
+
+		let package = SortedTransactions::sort_parents_child_package_topologically(txs);
+
+		assert_eq!(package.len(), 1);
+		assert_eq!(package[0].compute_txid(), parent_txid);
+	}
+
+	#[test]
+	fn topological_sort_accepts_empty_vec() {
+		SortedTransactions::sort_parents_child_package_topologically(Vec::new());
+	}
+}