Finish async kv store migration

examples/cli/src/main.rs

@@ -474,7 +474,7 @@ async fn execute_command(command: Commands, state: &mut WalletState) -> Result<(
 			);
 			println!(
 				"Rebalance Enabled: {}",
-				if wallet.get_rebalance_enabled() {
+				if wallet.get_rebalance_enabled().await {
 					"Yes".bright_green()
 				} else {
 					"No".bright_red()

orange-sdk/src/event.rs

@@ -339,7 +339,7 @@ impl LdkEventHandler {
 				let preimage = payment_preimage.unwrap(); // safe
 				let payment_id = PaymentId::SelfCustodial(payment_id.unwrap().0); // safe
 
-				if self.tx_metadata.set_preimage(payment_id, preimage.0).is_err() {
+				if self.tx_metadata.set_preimage(payment_id, preimage.0).await.is_err() {
 					log_error!(self.logger, "Failed to set preimage for payment {payment_id:?}");
 				}
 
@@ -442,7 +442,7 @@ impl LdkEventHandler {
 			} => {
 				// We experienced a channel close, we disable rebalancing so we don't automatically
 				// try to reopen the channel.
-				store::set_rebalance_enabled(self.event_queue.kv_store.as_ref(), false);
+				store::set_rebalance_enabled(self.event_queue.kv_store.as_ref(), false).await;
 
 				if let Err(e) = self
 					.event_queue

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

@@ -118,13 +118,13 @@ impl Wallet {
 	}
 
 	/// Sets whether the wallet should automatically rebalance from trusted/onchain to lightning.
-	pub fn set_rebalance_enabled(&self, value: bool) {
-		self.inner.set_rebalance_enabled(value)
+	pub async fn set_rebalance_enabled(&self, value: bool) {
+		self.inner.set_rebalance_enabled(value).await
 	}
 
 	/// Whether the wallet should automatically rebalance from trusted/onchain to lightning.
-	pub fn get_rebalance_enabled(&self) -> bool {
-		self.inner.get_rebalance_enabled()
+	pub async fn get_rebalance_enabled(&self) -> bool {
+		self.inner.get_rebalance_enabled().await
 	}
 
 	pub async fn list_transactions(
@@ -191,8 +191,8 @@ impl Wallet {
 	}
 
 	/// List our current channels
-	pub fn close_channels(&self) -> Result<(), WalletError> {
-		self.inner.close_channels()?;
+	pub async fn close_channels(&self) -> Result<(), WalletError> {
+		self.inner.close_channels().await?;
 		Ok(())
 	}
 

orange-sdk/src/lib.rs

@@ -634,13 +634,13 @@ impl Wallet {
 	}
 
 	/// Sets whether the wallet should automatically rebalance from trusted/onchain to lightning.
-	pub fn set_rebalance_enabled(&self, value: bool) {
-		store::set_rebalance_enabled(self.inner.store.as_ref(), value)
+	pub async fn set_rebalance_enabled(&self, value: bool) {
+		store::set_rebalance_enabled(self.inner.store.as_ref(), value).await
 	}
 
 	/// Whether the wallet should automatically rebalance from trusted/onchain to lightning.
-	pub fn get_rebalance_enabled(&self) -> bool {
-		store::get_rebalance_enabled(self.inner.store.as_ref())
+	pub async fn get_rebalance_enabled(&self) -> bool {
+		store::get_rebalance_enabled(self.inner.store.as_ref()).await
 	}
 
 	/// Returns the lightning wallet's node id.
@@ -664,7 +664,7 @@ impl Wallet {
 		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());
+		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(),
@@ -1095,15 +1095,18 @@ impl Wallet {
 					let res = self.inner.trusted.pay(method, instructions.amount).await;
 					match res {
 						Ok(id) => {
-							self.inner.tx_metadata.insert(
-								PaymentId::Trusted(id),
-								TxMetadata {
-									ty: TxType::Payment { ty: ty() },
-									time: SystemTime::now()
-										.duration_since(SystemTime::UNIX_EPOCH)
-										.unwrap(),
-								},
-							);
+							self.inner
+								.tx_metadata
+								.insert(
+									PaymentId::Trusted(id),
+									TxMetadata {
+										ty: TxType::Payment { ty: ty() },
+										time: SystemTime::now()
+											.duration_since(SystemTime::UNIX_EPOCH)
+											.unwrap(),
+									},
+								)
+								.await;
 							return Ok(PaymentId::Trusted(id));
 						},
 						Err(e) => {
@@ -1136,15 +1139,18 @@ impl Wallet {
 								// Note that the Payment Id can be repeated if we make a payment,
 								// it fails, then we attempt to pay the same (BOLT 11) invoice
 								// again.
-								self.inner.tx_metadata.upsert(
-									PaymentId::SelfCustodial(id.0),
-									TxMetadata {
-										ty: TxType::Payment { ty: typ },
-										time: SystemTime::now()
-											.duration_since(SystemTime::UNIX_EPOCH)
-											.unwrap(),
-									},
-								);
+								self.inner
+									.tx_metadata
+									.upsert(
+										PaymentId::SelfCustodial(id.0),
+										TxMetadata {
+											ty: TxType::Payment { ty: typ },
+											time: SystemTime::now()
+												.duration_since(SystemTime::UNIX_EPOCH)
+												.unwrap(),
+										},
+									)
+									.await;
 								let inner_ref = Arc::clone(&self.inner);
 								self.inner.runtime.spawn_cancellable_background_task(async move {
 									inner_ref.rebalancer.do_rebalance_if_needed().await;
@@ -1262,10 +1268,10 @@ impl Wallet {
 	/// Initiates closing all channels in the lightning wallet. The channel will not be closed
 	/// until a [`Event::ChannelClosed`] event is emitted.
 	/// This will disable rebalancing before closing channels, so that we don't try to reopen them.
-	pub fn close_channels(&self) -> Result<(), WalletError> {
+	pub async fn close_channels(&self) -> Result<(), WalletError> {
 		// we are explicitly disabling rebalancing here, so that we don't try to
 		// reopen channels after closing them.
-		self.set_rebalance_enabled(false);
+		self.set_rebalance_enabled(false).await;
 
 		self.inner.ln_wallet.close_channels()?;
 

orange-sdk/src/lightning_wallet.rs

@@ -366,7 +366,8 @@ impl LightningWallet {
 											store::write_splice_out(
 												self.inner.store.as_ref(),
 												&details,
-											);
+											)
+											.await;
 											return Ok(id);
 										}
 									},

orange-sdk/src/rebalancer.rs

@@ -62,7 +62,7 @@ impl OrangeTrigger {
 impl RebalanceTrigger for OrangeTrigger {
 	fn needs_trusted_rebalance(&self) -> impl Future<Output = Option<TriggerParams>> + Send {
 		async move {
-			let rebalance_enabled = store::get_rebalance_enabled(self.store.as_ref());
+			let rebalance_enabled = store::get_rebalance_enabled(self.store.as_ref()).await;
 			if !rebalance_enabled {
 				return None;
 			}
@@ -99,15 +99,17 @@ impl RebalanceTrigger for OrangeTrigger {
 							payment.id.as_hex()
 						);
 						new_txn.push((payment.amount, &payment.id));
-						self.tx_metadata.insert(
-							payment_id,
-							TxMetadata {
-								ty: TxType::Payment { ty: PaymentType::IncomingLightning {} },
-								time: SystemTime::now()
-									.duration_since(SystemTime::UNIX_EPOCH)
-									.unwrap(),
-							},
-						);
+						self.tx_metadata
+							.insert(
+								payment_id,
+								TxMetadata {
+									ty: TxType::Payment { ty: PaymentType::IncomingLightning {} },
+									time: SystemTime::now()
+										.duration_since(SystemTime::UNIX_EPOCH)
+										.unwrap(),
+								},
+							)
+							.await;
 					}
 				}
 
@@ -141,7 +143,7 @@ impl RebalanceTrigger for OrangeTrigger {
 
 	fn needs_onchain_rebalance(&self) -> impl Future<Output = Option<TriggerParams>> + Send {
 		async move {
-			let rebalance_enabled = store::get_rebalance_enabled(self.store.as_ref());
+			let rebalance_enabled = store::get_rebalance_enabled(self.store.as_ref()).await;
 			if !rebalance_enabled {
 				return None;
 			}
@@ -215,19 +217,21 @@ impl RebalanceTrigger for OrangeTrigger {
 								// make sure we have a metadata entry for the triggering transaction
 								let trigger = PaymentId::SelfCustodial(txid.to_byte_array());
 								if self.tx_metadata.read().get(&trigger).is_none() {
-									self.tx_metadata.insert(
-										trigger,
-										TxMetadata {
-											ty: TxType::Payment {
-												ty: PaymentType::IncomingOnChain {
-													txid: Some(txid),
+									self.tx_metadata
+										.insert(
+											trigger,
+											TxMetadata {
+												ty: TxType::Payment {
+													ty: PaymentType::IncomingOnChain {
+														txid: Some(txid),
+													},
 												},
+												time: SystemTime::now()
+													.duration_since(SystemTime::UNIX_EPOCH)
+													.unwrap(),
 											},
-											time: SystemTime::now()
-												.duration_since(SystemTime::UNIX_EPOCH)
-												.unwrap(),
-										},
-									);
+										)
+										.await;
 								}
 
 								Some(TriggerParams {
@@ -295,7 +299,8 @@ impl graduated_rebalancer::EventHandler for OrangeRebalanceEventHandler {
 						time: SystemTime::now().duration_since(SystemTime::UNIX_EPOCH).unwrap(),
 					};
 					self.tx_metadata
-						.insert(PaymentId::Trusted(trusted_rebalance_payment_id), metadata);
+						.insert(PaymentId::Trusted(trusted_rebalance_payment_id), metadata)
+						.await;
 					if let Err(e) = self
 						.event_queue
 						.add_event(Event::RebalanceInitiated {
@@ -318,6 +323,7 @@ impl graduated_rebalancer::EventHandler for OrangeRebalanceEventHandler {
 					let triggering_transaction_id = PaymentId::Trusted(trigger_id);
 					self.tx_metadata
 						.set_tx_caused_rebalance(&triggering_transaction_id)
+						.await
 						.expect("Failed to write metadata for rebalance transaction");
 					let metadata = TxMetadata {
 						ty: TxType::TrustedToLightning {
@@ -327,8 +333,8 @@ impl graduated_rebalancer::EventHandler for OrangeRebalanceEventHandler {
 						},
 						time: SystemTime::now().duration_since(SystemTime::UNIX_EPOCH).unwrap(),
 					};
-					self.tx_metadata.upsert(PaymentId::Trusted(rebalance_id), metadata);
-					self.tx_metadata.insert(PaymentId::SelfCustodial(lightning_id), metadata);
+					self.tx_metadata.upsert(PaymentId::Trusted(rebalance_id), metadata).await;
+					self.tx_metadata.insert(PaymentId::SelfCustodial(lightning_id), metadata).await;
 
 					let event_queue = Arc::clone(&self.event_queue);
 					let logger = Arc::clone(&self.logger);
@@ -357,13 +363,15 @@ impl graduated_rebalancer::EventHandler for OrangeRebalanceEventHandler {
 					let trigger_id = PaymentId::SelfCustodial(triggering_txid.to_byte_array());
 					self.tx_metadata
 						.set_tx_caused_rebalance(&trigger_id)
+						.await
 						.expect("Failed to write metadata for onchain rebalance transaction");
 					let metadata = TxMetadata {
 						ty: TxType::OnchainToLightning { channel_txid: chan_txid, triggering_txid },
 						time: SystemTime::now().duration_since(SystemTime::UNIX_EPOCH).unwrap(),
 					};
 					self.tx_metadata
-						.insert(PaymentId::SelfCustodial(chan_txid.to_byte_array()), metadata);
+						.insert(PaymentId::SelfCustodial(chan_txid.to_byte_array()), metadata)
+						.await;
 				},
 			}
 		})