Stop reorg nodes before runtime exit

Stop all reorg test nodes on helper threads before leaving each

property case so shutdown does not park the Tokio worker.

Co-Authored-By: HAL 9000

Author: tnull

tests/common/mod.rs

@@ -730,6 +730,43 @@ where
 	}
 }
 
+pub(crate) async fn stop_nodes(node_a: TestNode, node_b: TestNode) {
+	let (stop_a_sender, stop_a_receiver) = tokio::sync::oneshot::channel();
+	let (stop_b_sender, stop_b_receiver) = tokio::sync::oneshot::channel();
+	std::thread::spawn(move || {
+		let _ = stop_a_sender.send(node_a.stop());
+	});
+	std::thread::spawn(move || {
+		let _ = stop_b_sender.send(node_b.stop());
+	});
+	stop_a_receiver.await.expect("node_a stop thread panicked").unwrap();
+	stop_b_receiver.await.expect("node_b stop thread panicked").unwrap();
+}
+
+pub(crate) async fn stop_nodes_concurrently(nodes: Vec<TestNode>) {
+	let stop_receivers = nodes
+		.into_iter()
+		.map(|node| {
+			let (stop_sender, stop_receiver) = tokio::sync::oneshot::channel();
+			std::thread::spawn(move || {
+				let _ = stop_sender.send(node.stop());
+			});
+			stop_receiver
+		})
+		.collect::<Vec<_>>();
+
+	for stop_receiver in stop_receivers {
+		stop_receiver.await.expect("node stop thread panicked").unwrap();
+	}
+}
+
+pub(crate) async fn stop_node(node: TestNode) {
+	let (stop_sender, stop_receiver) = tokio::sync::oneshot::channel();
+	std::thread::spawn(move || {
+		let _ = stop_sender.send(node.stop());
+	});
+	stop_receiver.await.expect("node stop thread panicked").unwrap();
+}
 pub(crate) async fn premine_and_distribute_funds<E: ElectrumApi>(
 	bitcoind: &BitcoindClient, electrs: &E, addrs: Vec<Address>, amount: Amount,
 ) {

tests/reorg_test.rs

@@ -11,7 +11,7 @@ use proptest::proptest;
 use crate::common::{
 	expect_event, generate_blocks_and_wait, invalidate_blocks, open_channel,
 	premine_and_distribute_funds, random_chain_source, random_config, setup_bitcoind_and_electrsd,
-	setup_node, wait_for_outpoint_spend,
+	setup_node, stop_nodes_concurrently, wait_for_outpoint_spend,
 };
 
 async fn wait_for_pending_sweep_balance<F>(node: &ldk_node::Node, mut matches_pending_balance: F)
@@ -236,6 +236,8 @@ proptest! {
 
 				assert_eq!(node.next_event(), None);
 			});
+
+			stop_nodes_concurrently(nodes).await;
 		})
 	}
 }