Fixup changes

Author: benthecarman

orange-sdk/src/event.rs

@@ -14,7 +14,7 @@ use ldk_node::payment::{ConfirmationStatus, PaymentKind};
 use ldk_node::{CustomTlvRecord, DynStore, UserChannelId};
 
 use std::collections::VecDeque;
-use std::sync::{Arc, Condvar, Mutex};
+use std::sync::{Arc, Mutex};
 use std::task::{Poll, Waker};
 use tokio::sync::watch;
 
@@ -190,7 +190,6 @@ impl_writeable_tlv_based_enum!(Event,
 pub struct EventQueue {
 	queue: Arc<Mutex<VecDeque<Event>>>,
 	waker: Arc<Mutex<Option<Waker>>>,
-	notifier: Condvar,
 	kv_store: Arc<DynStore>,
 	logger: Arc<Logger>,
 }
@@ -199,8 +198,7 @@ impl EventQueue {
 	pub(crate) fn new(kv_store: Arc<DynStore>, logger: Arc<Logger>) -> Self {
 		let queue = Arc::new(Mutex::new(VecDeque::new()));
 		let waker = Arc::new(Mutex::new(None));
-		let notifier = Condvar::new();
-		Self { queue, waker, notifier, kv_store, logger }
+		Self { queue, waker, kv_store, logger }
 	}
 
 	pub(crate) fn add_event(&self, event: Event) -> Result<(), ldk_node::lightning::io::Error> {
@@ -210,8 +208,6 @@ impl EventQueue {
 			self.persist_queue(&locked_queue)?;
 		}
 
-		self.notifier.notify_one();
-
 		if let Some(waker) = self.waker.lock().unwrap().take() {
 			waker.wake();
 		}
@@ -227,19 +223,12 @@ impl EventQueue {
 		EventFuture { event_queue: Arc::clone(&self.queue), waker: Arc::clone(&self.waker) }.await
 	}
 
-	pub(crate) fn wait_next_event(&self) -> Event {
-		let locked_queue =
-			self.notifier.wait_while(self.queue.lock().unwrap(), |queue| queue.is_empty()).unwrap();
-		locked_queue.front().unwrap().clone()
-	}
-
 	pub(crate) fn event_handled(&self) -> Result<(), ldk_node::lightning::io::Error> {
 		{
 			let mut locked_queue = self.queue.lock().unwrap();
 			locked_queue.pop_front();
 			self.persist_queue(&locked_queue)?;
 		}
-		self.notifier.notify_one();
 
 		if let Some(waker) = self.waker.lock().unwrap().take() {
 			waker.wake();

orange-sdk/src/ffi/bitcoin_payment_instructions.rs

@@ -85,6 +85,7 @@ pub enum ParseError {
 	UnknownRequiredParameter,
 	HrnResolutionError(String),
 	InstructionsExpired,
+	InvalidLnurl(String),
 }
 
 impl Display for ParseError {
@@ -108,6 +109,7 @@ impl Display for ParseError {
 				write!(f, "Human readable name resolution error: {}", e)
 			},
 			ParseError::InstructionsExpired => write!(f, "Payment instructions have expired"),
+			ParseError::InvalidLnurl(e) => write!(f, "Invalid LNURL: {}", e),
 		}
 	}
 }
@@ -131,6 +133,7 @@ impl From<BPIParseError> for ParseError {
 				ParseError::HrnResolutionError(msg.to_string())
 			},
 			BPIParseError::InstructionsExpired => ParseError::InstructionsExpired,
+			BPIParseError::InvalidLnurl(msg) => ParseError::InvalidLnurl(msg.to_string()),
 		}
 	}
 }

orange-sdk/src/lib.rs

@@ -1304,7 +1304,10 @@ impl Wallet {
 	/// **Caution:** Users must handle events as quickly as possible to prevent a large event backlog,
 	/// which can increase the memory footprint of [`Wallet`].
 	pub fn wait_next_event(&self) -> Event {
-		self.inner.event_queue.wait_next_event()
+		let fut = self.inner.event_queue.next_event_async();
+		// We use our runtime for the sync variant to ensure `tokio::task::block_in_place` is
+		// always called if we'd ever hit this in an outer runtime context.
+		self.inner.runtime.block_on(fut)
 	}
 
 	/// Confirm the last retrieved event handled.

orange-sdk/tests/integration_tests.rs

@@ -18,7 +18,7 @@ use std::time::Duration;
 
 mod test_utils;
 
-#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
+#[tokio::test(flavor = "multi_thread")]
 async fn test_node_start() {
 	let TestParams { wallet, .. } = build_test_nodes().await;
 
@@ -88,7 +88,7 @@ async fn test_receive_to_trusted() {
 	lsp.stop().unwrap();
 }
 
-#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
+#[tokio::test(flavor = "multi_thread")]
 async fn test_pay_from_trusted() {
 	let TestParams { wallet, third_party, lsp, .. } = build_test_nodes().await;
 
@@ -167,7 +167,7 @@ async fn test_pay_from_trusted() {
 	}
 }
 
-#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
+#[tokio::test(flavor = "multi_thread")]
 async fn test_sweep_to_ln() {
 	let TestParams { wallet, lsp, third_party, .. } = build_test_nodes().await;
 
@@ -320,7 +320,7 @@ async fn test_sweep_to_ln() {
 	);
 }
 
-#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
+#[tokio::test(flavor = "multi_thread")]
 async fn test_receive_to_ln() {
 	let TestParams { wallet, third_party, .. } = build_test_nodes().await;
 
@@ -358,7 +358,7 @@ async fn test_receive_to_ln() {
 	);
 }
 
-#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
+#[tokio::test(flavor = "multi_thread")]
 async fn test_receive_to_onchain() {
 	let TestParams { wallet, lsp, bitcoind, third_party, .. } = build_test_nodes().await;
 
@@ -556,7 +556,7 @@ async fn run_test_pay_lightning_from_self_custody(amountless: bool) {
 		&& p.amount_msat == Some(amount.milli_sats())));
 }
 
-#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
+#[tokio::test(flavor = "multi_thread")]
 async fn test_pay_lightning_from_self_custody() {
 	run_test_pay_lightning_from_self_custody(false).await;
 	run_test_pay_lightning_from_self_custody(true).await;
@@ -644,13 +644,13 @@ async fn run_test_pay_bolt12_from_self_custody(amountless: bool) {
 		&& p.amount_msat == Some(amount.milli_sats())));
 }
 
-#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
+#[tokio::test(flavor = "multi_thread")]
 async fn test_pay_bolt12_from_self_custody() {
 	run_test_pay_bolt12_from_self_custody(false).await;
 	run_test_pay_bolt12_from_self_custody(true).await;
 }
 
-#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
+#[tokio::test(flavor = "multi_thread")]
 async fn test_pay_onchain_from_self_custody() {
 	let TestParams { wallet, bitcoind, third_party, .. } = build_test_nodes().await;
 
@@ -753,7 +753,7 @@ async fn test_pay_onchain_from_self_custody() {
 	.await;
 }
 
-#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
+#[tokio::test(flavor = "multi_thread")]
 async fn test_force_close_handling() {
 	let TestParams { wallet, lsp, bitcoind, third_party, .. } = build_test_nodes().await;
 
@@ -794,7 +794,7 @@ async fn test_force_close_handling() {
 	assert!(!rebalancing);
 }
 
-#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
+#[tokio::test(flavor = "multi_thread")]
 async fn test_close_all_channels() {
 	let TestParams { wallet, lsp, bitcoind, third_party, .. } = build_test_nodes().await;
 
@@ -828,7 +828,7 @@ async fn test_close_all_channels() {
 	assert!(!rebalancing);
 }
 
-#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
+#[tokio::test(flavor = "multi_thread")]
 async fn test_threshold_boundary_trusted_balance_limit() {
 	let TestParams { wallet, third_party, .. } = build_test_nodes().await;
 
@@ -904,7 +904,7 @@ async fn test_threshold_boundary_trusted_balance_limit() {
 	);
 }
 
-#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
+#[tokio::test(flavor = "multi_thread")]
 async fn test_threshold_boundary_rebalance_min() {
 	let TestParams { wallet, third_party, .. } = build_test_nodes().await;
 
@@ -977,7 +977,7 @@ async fn test_threshold_boundary_rebalance_min() {
 	);
 }
 
-#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
+#[tokio::test(flavor = "multi_thread")]
 async fn test_threshold_boundary_onchain_receive_threshold() {
 	let TestParams { wallet, .. } = build_test_nodes().await;
 
@@ -1037,7 +1037,7 @@ async fn test_threshold_boundary_onchain_receive_threshold() {
 	}
 }
 
-#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
+#[tokio::test(flavor = "multi_thread")]
 async fn test_threshold_combinations_and_edge_cases() {
 	let TestParams { wallet, .. } = build_test_nodes().await;
 
@@ -1095,7 +1095,7 @@ async fn test_threshold_combinations_and_edge_cases() {
 	}
 }
 
-#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
+#[tokio::test(flavor = "multi_thread")]
 async fn test_invalid_payment_instructions() {
 	let TestParams { wallet, third_party, .. } = build_test_nodes().await;
 
@@ -1156,7 +1156,7 @@ async fn test_invalid_payment_instructions() {
 	assert_eq!(txs.len(), 0, "Failed payments should not be recorded in transaction list");
 }
 
-#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
+#[tokio::test(flavor = "multi_thread")]
 async fn test_payment_with_expired_invoice() {
 	let TestParams { wallet, third_party, .. } = build_test_nodes().await;
 
@@ -1185,7 +1185,7 @@ async fn test_payment_with_expired_invoice() {
 	assert!(matches!(parse_result.unwrap_err(), ParseError::InstructionsExpired));
 }
 
-#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
+#[tokio::test(flavor = "multi_thread")]
 async fn test_payment_network_mismatch() {
 	let TestParams { wallet, bitcoind, .. } = build_test_nodes().await;
 
@@ -1238,7 +1238,7 @@ async fn test_payment_network_mismatch() {
 	);
 }
 
-#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
+#[tokio::test(flavor = "multi_thread")]
 async fn test_concurrent_payments() {
 	let TestParams { wallet, bitcoind, third_party, .. } = build_test_nodes().await;
 
@@ -1410,7 +1410,7 @@ async fn test_concurrent_payments() {
 	);
 }
 
-#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
+#[tokio::test(flavor = "multi_thread")]
 async fn test_concurrent_receive_operations() {
 	let TestParams { wallet, third_party, .. } = build_test_nodes().await;
 
@@ -1465,7 +1465,7 @@ async fn test_concurrent_receive_operations() {
 	assert_eq!(incoming_count, 2, "Should have exactly 2 incoming transactions");
 }
 
-#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
+#[tokio::test(flavor = "multi_thread")]
 async fn test_balance_consistency_under_load() {
 	let TestParams { wallet, third_party, .. } = build_test_nodes().await;
 
@@ -1516,7 +1516,7 @@ async fn test_balance_consistency_under_load() {
 	}
 }
 
-#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
+#[tokio::test(flavor = "multi_thread")]
 async fn test_invalid_tunables_relationships() {
 	let TestParams { wallet, .. } = build_test_nodes().await;
 
@@ -1593,7 +1593,7 @@ async fn test_invalid_tunables_relationships() {
 	}
 }
 
-#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
+#[tokio::test(flavor = "multi_thread")]
 async fn test_extreme_amount_handling() {
 	let TestParams { wallet, .. } = build_test_nodes().await;
 
@@ -1658,7 +1658,7 @@ async fn test_extreme_amount_handling() {
 	// On-chain address depends on threshold, not msat precision
 }
 
-#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
+#[tokio::test(flavor = "multi_thread")]
 async fn test_wallet_configuration_validation() {
 	let TestParams { wallet, .. } = build_test_nodes().await;
 
@@ -1721,7 +1721,7 @@ async fn test_wallet_configuration_validation() {
 	);
 }
 
-#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
+#[tokio::test(flavor = "multi_thread")]
 async fn test_edge_case_payment_instruction_parsing() {
 	let TestParams { wallet, third_party, .. } = build_test_nodes().await;
 
@@ -1771,7 +1771,7 @@ async fn test_edge_case_payment_instruction_parsing() {
 	}
 }
 
-#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
+#[tokio::test(flavor = "multi_thread")]
 async fn test_lsp_connectivity_fallback() {
 	let TestParams { wallet, lsp, bitcoind, third_party, .. } = build_test_nodes().await;
 

orange-sdk/tests/test_utils.rs

@@ -64,7 +64,7 @@ pub async fn wait_next_event(wallet: &orange_sdk::Wallet) -> orange_sdk::Event {
 fn create_bitcoind(uuid: Uuid) -> Bitcoind {
 	let mut conf = Conf::default();
 	conf.args.push("-txindex");
-	conf.args.push("-rpcworkqueue=100");
+	conf.args.push("-rpcworkqueue=200");
 	conf.staticdir = Some(temp_dir().join(format!("orange-test-{uuid}/bitcoind")));
 	let bitcoind = Bitcoind::with_conf(corepc_node::downloaded_exe_path().unwrap(), &conf)
 		.unwrap_or_else(|_| panic!("Failed to start bitcoind for test {uuid}"));