Merge pull request #45 from TheBlueMatt-macbook/2025-12-async-opts

Author: benthecarman

orange-sdk/Cargo.toml

@@ -14,7 +14,7 @@ name = "orange_sdk"
 
 [features]
 default = ["spark"]
-uniffi = ["dep:uniffi", "spark", "cashu", "rand"]
+uniffi = ["dep:uniffi", "spark", "cashu", "rand", "pin-project-lite"]
 spark = ["breez-sdk-spark", "uuid", "serde_json"]
 cashu = ["cdk", "serde_json"]
 _test-utils = ["corepc-node", 'electrsd', "cashu", "uuid/v7", "rand"]
@@ -29,7 +29,7 @@ bitcoin-payment-instructions = { workspace = true, features = ["http"] }
 chrono = { version = "0.4", default-features = false }
 rand = { version = "0.8.5", optional = true }
 breez-sdk-spark = { git = "https://github.com/breez/spark-sdk.git", rev = "5f25eb6185950545e2ca170971667adc43c3544f", default-features = false, features = ["rustls-tls"], optional = true }
-tokio = { version = "1.0", default-features = false, features = ["rt-multi-thread", "sync"] }
+tokio = { version = "1.0", default-features = false, features = ["rt-multi-thread", "sync", "macros"] }
 uuid = { version = "1.0", default-features = false, optional = true }
 cdk = { version = "0.14.2", default-features = false, features = ["wallet"], optional = true }
 serde_json = { version = "1.0", optional = true }
@@ -43,7 +43,8 @@ cdk-sqlite = { version = "0.14.2", optional = true }
 cdk-axum = { version = "0.14.2", optional = true }
 axum = { version = "0.8.1", optional = true }
 
-uniffi = { version = "0.29", features = ["cli", "tokio"], optional = true }
+uniffi = { version = "0.29", default-features = false, features = ["cli"], optional = true }
+pin-project-lite = { version = "0.2", default-features = false, optional = true }
 
 [dev-dependencies]
 test-log = "0.2.18"

orange-sdk/src/ffi/orange/wallet.rs

@@ -10,7 +10,10 @@ use crate::ffi::orange::Event;
 use crate::ffi::orange::config::{Tunables, WalletConfig};
 use crate::ffi::orange::error::{InitFailure, WalletError};
 use crate::{impl_from_core_type, impl_into_core_type};
+
+use std::pin::Pin;
 use std::sync::Arc;
+use std::task::{Context, Poll};
 
 /// Represents the balances of the wallet, including available and pending balances.
 #[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, uniffi::Object)]
@@ -85,52 +88,82 @@ impl SingleUseReceiveUri {
 impl_from_core_type!(OrangeSingleUseReceiveUri, SingleUseReceiveUri);
 impl_into_core_type!(SingleUseReceiveUri, OrangeSingleUseReceiveUri);
 
+// This is basically the `async-compat` utility that UniFFI uses under
+// the hood for its `async_runtime = "tokio"` attribute, except it lets
+// us use our own runtime to avoid the redundant runtimes implicit in
+// the `async-compat` single-threaded runtime.
+pin_project_lite::pin_project! {
+	struct RTPoller<T> {
+		#[pin]
+		inner: T,
+		rt: Arc<tokio::runtime::Runtime>,
+	}
+}
+
+impl<T> RTPoller<T> {
+	fn new(inner: T, rt: Arc<tokio::runtime::Runtime>) -> Self {
+		Self { inner, rt }
+	}
+}
+
+impl<T: Future> Future for RTPoller<T> {
+	type Output = T::Output;
+
+	fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+		let _guard = self.rt.enter();
+		self.project().inner.poll(cx)
+	}
+}
+
 #[derive(Clone, uniffi::Object)]
 pub struct Wallet {
 	inner: Arc<OrangeWallet>,
-	_rt: Arc<tokio::runtime::Runtime>,
+	rt: Arc<tokio::runtime::Runtime>,
 }
 
-#[uniffi::export(async_runtime = "tokio")]
+#[uniffi::export]
 impl Wallet {
 	#[uniffi::constructor]
-	pub fn new(config: WalletConfig) -> Result<Self, InitFailure> {
-		let rt = Arc::new(tokio::runtime::Builder::new_multi_thread().enable_all().build()?);
+	pub async fn new(config: WalletConfig) -> Result<Self, InitFailure> {
+		let rt = Arc::new(
+			tokio::runtime::Builder::new_multi_thread().worker_threads(2).enable_all().build()?,
+		);
 
 		let config: OrangeWalletConfig = config.try_into()?;
 
-		let inner = rt.block_on(async move { OrangeWallet::new(config).await })?;
+		let inner = RTPoller::new(OrangeWallet::new(config), Arc::clone(&rt)).await?;
 
-		Ok(Wallet { inner: Arc::new(inner), _rt: rt })
+		Ok(Wallet { inner: Arc::new(inner), rt })
 	}
 
 	pub fn node_id(&self) -> String {
 		self.inner.node_id().to_string()
 	}
 
 	pub async fn get_balance(&self) -> Result<Balances, WalletError> {
-		let balance = self.inner.get_balance().await?;
+		let balance = RTPoller::new(self.inner.get_balance(), Arc::clone(&self.rt)).await?;
 		Ok(balance.into())
 	}
 
-	pub async fn is_connected_to_lsp(&self) -> bool {
+	pub fn is_connected_to_lsp(&self) -> bool {
 		self.inner.is_connected_to_lsp()
 	}
 
 	/// Sets whether the wallet should automatically rebalance from trusted/onchain to lightning.
 	pub async fn set_rebalance_enabled(&self, value: bool) {
-		self.inner.set_rebalance_enabled(value).await
+		RTPoller::new(self.inner.set_rebalance_enabled(value), Arc::clone(&self.rt)).await
 	}
 
 	/// Whether the wallet should automatically rebalance from trusted/onchain to lightning.
 	pub async fn get_rebalance_enabled(&self) -> bool {
-		self.inner.get_rebalance_enabled().await
+		RTPoller::new(self.inner.get_rebalance_enabled(), Arc::clone(&self.rt)).await
 	}
 
 	pub async fn list_transactions(
 		&self,
 	) -> Result<Vec<std::sync::Arc<crate::ffi::orange::Transaction>>, WalletError> {
-		let transactions = self.inner.list_transactions().await?;
+		let transactions =
+			RTPoller::new(self.inner.list_transactions(), Arc::clone(&self.rt)).await?;
 		Ok(transactions.into_iter().map(|tx| std::sync::Arc::new(tx.into())).collect())
 	}
 
@@ -144,7 +177,11 @@ impl Wallet {
 	pub async fn parse_payment_instructions(
 		&self, instructions: String,
 	) -> Result<PaymentInstructions, ParseError> {
-		let result = self.inner.parse_payment_instructions(&instructions).await?;
+		let result = RTPoller::new(
+			self.inner.parse_payment_instructions(&instructions),
+			Arc::clone(&self.rt),
+		)
+		.await?;
 		Ok(result.into())
 	}
 
@@ -159,7 +196,7 @@ impl Wallet {
 	pub async fn pay(
 		&self, payment_info: Arc<PaymentInfo>,
 	) -> Result<super::PaymentId, WalletError> {
-		let id = self.inner.pay(&payment_info.0).await?;
+		let id = RTPoller::new(self.inner.pay(&payment_info.0), Arc::clone(&self.rt)).await?;
 		Ok(id.into())
 	}
 
@@ -170,18 +207,26 @@ impl Wallet {
 	pub async fn estimate_fee(
 		&self, payment_info: Arc<PaymentInfo>,
 	) -> Result<Arc<Amount>, WalletError> {
-		let fee = self.inner.estimate_fee(&payment_info.0.instructions).await;
+		let fee = RTPoller::new(
+			self.inner.estimate_fee(&payment_info.0.instructions),
+			Arc::clone(&self.rt),
+		)
+		.await;
 		Ok(Arc::new(fee.into()))
 	}
 
 	pub async fn stop(&self) {
-		self.inner.stop().await
+		RTPoller::new(self.inner.stop(), Arc::clone(&self.rt)).await
 	}
 
 	pub async fn get_single_use_receive_uri(
 		&self, amount: Option<Arc<Amount>>,
 	) -> Result<SingleUseReceiveUri, WalletError> {
-		let uri = self.inner.get_single_use_receive_uri(amount.map(|a| a.0)).await?;
+		let uri = RTPoller::new(
+			self.inner.get_single_use_receive_uri(amount.map(|a| a.0)),
+			Arc::clone(&self.rt),
+		)
+		.await?;
 		Ok(uri.into())
 	}
 
@@ -192,7 +237,7 @@ impl Wallet {
 
 	/// List our current channels
 	pub async fn close_channels(&self) -> Result<(), WalletError> {
-		self.inner.close_channels().await?;
+		RTPoller::new(self.inner.close_channels(), Arc::clone(&self.rt)).await?;
 		Ok(())
 	}
 
@@ -230,7 +275,7 @@ impl Wallet {
 	/// **Caution:** Users must handle events as quickly as possible to prevent a large event backlog,
 	/// which can increase the memory footprint of [`crate::Wallet`].
 	pub async fn next_event_async(&self) -> Event {
-		self.inner.next_event_async().await.into()
+		RTPoller::new(self.inner.next_event_async(), Arc::clone(&self.rt)).await.into()
 	}
 
 	/// Returns the next event in the event queue.

orange-sdk/src/lib.rs

@@ -208,6 +208,7 @@ pub enum ChainSource {
 }
 
 /// Configuration for initializing the wallet.
+#[derive(Clone)]
 pub struct WalletConfig {
 	/// Configuration for wallet storage.
 	pub storage_config: StorageConfig,
@@ -528,58 +529,66 @@ impl Wallet {
 
 		let tx_metadata = TxMetadataStore::new(Arc::clone(&store)).await;
 
-		let trusted: Arc<Box<DynTrustedWalletInterface>> = match &config.extra_config {
-			#[cfg(feature = "spark")]
-			ExtraConfig::Spark(sp) => Arc::new(Box::new(
-				Spark::init(
-					&config,
-					*sp,
-					Arc::clone(&store),
-					Arc::clone(&event_queue),
-					tx_metadata.clone(),
-					Arc::clone(&logger),
-					Arc::clone(&runtime),
-				)
-				.await?,
-			)),
-			#[cfg(feature = "cashu")]
-			ExtraConfig::Cashu(cashu) => Arc::new(Box::new(
-				Cashu::init(
-					&config,
-					cashu.clone(),
-					Arc::clone(&store),
-					Arc::clone(&event_queue),
-					tx_metadata.clone(),
-					Arc::clone(&logger),
-					Arc::clone(&runtime),
-				)
-				.await?,
-			)),
-			#[cfg(feature = "_test-utils")]
-			ExtraConfig::Dummy(cfg) => Arc::new(Box::new(
-				DummyTrustedWallet::new(
-					cfg.uuid,
-					&cfg.lsp,
-					&cfg.bitcoind,
-					tx_metadata.clone(),
-					Arc::clone(&event_queue),
-					Arc::clone(&runtime),
-				)
-				.await,
-			)),
-		};
-
-		let ln_wallet = Arc::new(
-			LightningWallet::init(
-				Arc::clone(&runtime),
-				config,
-				Arc::clone(&store),
-				Arc::clone(&event_queue),
-				tx_metadata.clone(),
-				Arc::clone(&logger),
-			)
-			.await?,
+		let (trusted, ln_wallet) = tokio::join!(
+			async {
+				let trusted: Arc<Box<DynTrustedWalletInterface>> = match &config.extra_config {
+					#[cfg(feature = "spark")]
+					ExtraConfig::Spark(sp) => Arc::new(Box::new(
+						Spark::init(
+							&config,
+							*sp,
+							Arc::clone(&store),
+							Arc::clone(&event_queue),
+							tx_metadata.clone(),
+							Arc::clone(&logger),
+							Arc::clone(&runtime),
+						)
+						.await?,
+					)),
+					#[cfg(feature = "cashu")]
+					ExtraConfig::Cashu(cashu) => Arc::new(Box::new(
+						Cashu::init(
+							&config,
+							cashu.clone(),
+							Arc::clone(&store),
+							Arc::clone(&event_queue),
+							tx_metadata.clone(),
+							Arc::clone(&logger),
+							Arc::clone(&runtime),
+						)
+						.await?,
+					)),
+					#[cfg(feature = "_test-utils")]
+					ExtraConfig::Dummy(cfg) => Arc::new(Box::new(
+						DummyTrustedWallet::new(
+							cfg.uuid,
+							&cfg.lsp,
+							&cfg.bitcoind,
+							tx_metadata.clone(),
+							Arc::clone(&event_queue),
+							Arc::clone(&runtime),
+						)
+						.await,
+					)),
+				};
+				Ok::<_, InitFailure>(trusted)
+			},
+			async {
+				Ok::<_, InitFailure>(Arc::new(
+					LightningWallet::init(
+						Arc::clone(&runtime),
+						config.clone(),
+						Arc::clone(&store),
+						Arc::clone(&event_queue),
+						tx_metadata.clone(),
+						Arc::clone(&logger),
+					)
+					.await?,
+				))
+			},
 		);
+		let trusted = trusted?;
+		let ln_wallet = ln_wallet?;
 
 		let wt = Arc::new(WalletTrusted(Arc::clone(&trusted)));
 
@@ -669,12 +678,14 @@ impl Wallet {
 
 	/// Lists the transactions which have been made.
 	pub async fn list_transactions(&self) -> Result<Vec<Transaction>, WalletError> {
-		let trusted_payments = self.inner.trusted.list_payments().await?;
+		let (trusted_payments, splice_outs) = tokio::join!(
+			self.inner.trusted.list_payments(),
+			store::read_splice_outs(self.inner.store.as_ref())
+		);
+		let trusted_payments = trusted_payments?;
 		let mut lightning_payments = self.inner.ln_wallet.list_payments();
 		lightning_payments.sort_by_key(|l| l.latest_update_timestamp);
 
-		let splice_outs = store::read_splice_outs(self.inner.store.as_ref()).await;
-
 		let mut res = Vec::with_capacity(
 			trusted_payments.len() + lightning_payments.len() + splice_outs.len(),
 		);