Skip to content

Commit e83dcca

Browse files
benthecarmanbenthecarman
authored
Finish async kv store migration (#44)
1 parent c03d831 commit e83dcca

11 files changed

Lines changed: 192 additions & 186 deletions

File tree

‎examples/cli/src/main.rs‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -474,7 +474,7 @@ async fn execute_command(command: Commands, state: &mut WalletState) -> Result<(
474474
);
475475
println!(
476476
"Rebalance Enabled: {}",
477-
if wallet.get_rebalance_enabled() {
477+
if wallet.get_rebalance_enabled().await {
478478
"Yes".bright_green()
479479
} else {
480480
"No".bright_red()

‎orange-sdk/src/event.rs‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -339,7 +339,7 @@ impl LdkEventHandler {
339339
let preimage = payment_preimage.unwrap(); // safe
340340
let payment_id = PaymentId::SelfCustodial(payment_id.unwrap().0); // safe
341341

342-
if self.tx_metadata.set_preimage(payment_id, preimage.0).is_err() {
342+
if self.tx_metadata.set_preimage(payment_id, preimage.0).await.is_err() {
343343
log_error!(self.logger, "Failed to set preimage for payment {payment_id:?}");
344344
}
345345

@@ -442,7 +442,7 @@ impl LdkEventHandler {
442442
} => {
443443
// We experienced a channel close, we disable rebalancing so we don't automatically
444444
// try to reopen the channel.
445-
store::set_rebalance_enabled(self.event_queue.kv_store.as_ref(), false);
445+
store::set_rebalance_enabled(self.event_queue.kv_store.as_ref(), false).await;
446446

447447
if let Err(e) = self
448448
.event_queue

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

Lines changed: 6 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -118,13 +118,13 @@ impl Wallet {
118118
}
119119

120120
/// Sets whether the wallet should automatically rebalance from trusted/onchain to lightning.
121-
pub fn set_rebalance_enabled(&self, value: bool) {
122-
self.inner.set_rebalance_enabled(value)
121+
pub async fn set_rebalance_enabled(&self, value: bool) {
122+
self.inner.set_rebalance_enabled(value).await
123123
}
124124

125125
/// Whether the wallet should automatically rebalance from trusted/onchain to lightning.
126-
pub fn get_rebalance_enabled(&self) -> bool {
127-
self.inner.get_rebalance_enabled()
126+
pub async fn get_rebalance_enabled(&self) -> bool {
127+
self.inner.get_rebalance_enabled().await
128128
}
129129

130130
pub async fn list_transactions(
@@ -191,8 +191,8 @@ impl Wallet {
191191
}
192192

193193
/// List our current channels
194-
pub fn close_channels(&self) -> Result<(), WalletError> {
195-
self.inner.close_channels()?;
194+
pub async fn close_channels(&self) -> Result<(), WalletError> {
195+
self.inner.close_channels().await?;
196196
Ok(())
197197
}
198198

‎orange-sdk/src/lib.rs‎

Lines changed: 31 additions & 25 deletions
Original file line numberDiff line numberDiff line change
@@ -634,13 +634,13 @@ impl Wallet {
634634
}
635635

636636
/// Sets whether the wallet should automatically rebalance from trusted/onchain to lightning.
637-
pub fn set_rebalance_enabled(&self, value: bool) {
638-
store::set_rebalance_enabled(self.inner.store.as_ref(), value)
637+
pub async fn set_rebalance_enabled(&self, value: bool) {
638+
store::set_rebalance_enabled(self.inner.store.as_ref(), value).await
639639
}
640640

641641
/// Whether the wallet should automatically rebalance from trusted/onchain to lightning.
642-
pub fn get_rebalance_enabled(&self) -> bool {
643-
store::get_rebalance_enabled(self.inner.store.as_ref())
642+
pub async fn get_rebalance_enabled(&self) -> bool {
643+
store::get_rebalance_enabled(self.inner.store.as_ref()).await
644644
}
645645

646646
/// Returns the lightning wallet's node id.
@@ -664,7 +664,7 @@ impl Wallet {
664664
let mut lightning_payments = self.inner.ln_wallet.list_payments();
665665
lightning_payments.sort_by_key(|l| l.latest_update_timestamp);
666666

667-
let splice_outs = store::read_splice_outs(self.inner.store.as_ref());
667+
let splice_outs = store::read_splice_outs(self.inner.store.as_ref()).await;
668668

669669
let mut res = Vec::with_capacity(
670670
trusted_payments.len() + lightning_payments.len() + splice_outs.len(),
@@ -1095,15 +1095,18 @@ impl Wallet {
10951095
let res = self.inner.trusted.pay(method, instructions.amount).await;
10961096
match res {
10971097
Ok(id) => {
1098-
self.inner.tx_metadata.insert(
1099-
PaymentId::Trusted(id),
1100-
TxMetadata {
1101-
ty: TxType::Payment { ty: ty() },
1102-
time: SystemTime::now()
1103-
.duration_since(SystemTime::UNIX_EPOCH)
1104-
.unwrap(),
1105-
},
1106-
);
1098+
self.inner
1099+
.tx_metadata
1100+
.insert(
1101+
PaymentId::Trusted(id),
1102+
TxMetadata {
1103+
ty: TxType::Payment { ty: ty() },
1104+
time: SystemTime::now()
1105+
.duration_since(SystemTime::UNIX_EPOCH)
1106+
.unwrap(),
1107+
},
1108+
)
1109+
.await;
11071110
return Ok(PaymentId::Trusted(id));
11081111
},
11091112
Err(e) => {
@@ -1136,15 +1139,18 @@ impl Wallet {
11361139
// Note that the Payment Id can be repeated if we make a payment,
11371140
// it fails, then we attempt to pay the same (BOLT 11) invoice
11381141
// again.
1139-
self.inner.tx_metadata.upsert(
1140-
PaymentId::SelfCustodial(id.0),
1141-
TxMetadata {
1142-
ty: TxType::Payment { ty: typ },
1143-
time: SystemTime::now()
1144-
.duration_since(SystemTime::UNIX_EPOCH)
1145-
.unwrap(),
1146-
},
1147-
);
1142+
self.inner
1143+
.tx_metadata
1144+
.upsert(
1145+
PaymentId::SelfCustodial(id.0),
1146+
TxMetadata {
1147+
ty: TxType::Payment { ty: typ },
1148+
time: SystemTime::now()
1149+
.duration_since(SystemTime::UNIX_EPOCH)
1150+
.unwrap(),
1151+
},
1152+
)
1153+
.await;
11481154
let inner_ref = Arc::clone(&self.inner);
11491155
self.inner.runtime.spawn_cancellable_background_task(async move {
11501156
inner_ref.rebalancer.do_rebalance_if_needed().await;
@@ -1262,10 +1268,10 @@ impl Wallet {
12621268
/// Initiates closing all channels in the lightning wallet. The channel will not be closed
12631269
/// until a [`Event::ChannelClosed`] event is emitted.
12641270
/// This will disable rebalancing before closing channels, so that we don't try to reopen them.
1265-
pub fn close_channels(&self) -> Result<(), WalletError> {
1271+
pub async fn close_channels(&self) -> Result<(), WalletError> {
12661272
// we are explicitly disabling rebalancing here, so that we don't try to
12671273
// reopen channels after closing them.
1268-
self.set_rebalance_enabled(false);
1274+
self.set_rebalance_enabled(false).await;
12691275

12701276
self.inner.ln_wallet.close_channels()?;
12711277

‎orange-sdk/src/lightning_wallet.rs‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -366,7 +366,8 @@ impl LightningWallet {
366366
store::write_splice_out(
367367
self.inner.store.as_ref(),
368368
&details,
369-
);
369+
)
370+
.await;
370371
return Ok(id);
371372
}
372373
},

‎orange-sdk/src/rebalancer.rs‎

Lines changed: 34 additions & 26 deletions
Original file line numberDiff line numberDiff line change
@@ -62,7 +62,7 @@ impl OrangeTrigger {
6262
impl RebalanceTrigger for OrangeTrigger {
6363
fn needs_trusted_rebalance(&self) -> impl Future<Output = Option<TriggerParams>> + Send {
6464
async move {
65-
let rebalance_enabled = store::get_rebalance_enabled(self.store.as_ref());
65+
let rebalance_enabled = store::get_rebalance_enabled(self.store.as_ref()).await;
6666
if !rebalance_enabled {
6767
return None;
6868
}
@@ -99,15 +99,17 @@ impl RebalanceTrigger for OrangeTrigger {
9999
payment.id.as_hex()
100100
);
101101
new_txn.push((payment.amount, &payment.id));
102-
self.tx_metadata.insert(
103-
payment_id,
104-
TxMetadata {
105-
ty: TxType::Payment { ty: PaymentType::IncomingLightning {} },
106-
time: SystemTime::now()
107-
.duration_since(SystemTime::UNIX_EPOCH)
108-
.unwrap(),
109-
},
110-
);
102+
self.tx_metadata
103+
.insert(
104+
payment_id,
105+
TxMetadata {
106+
ty: TxType::Payment { ty: PaymentType::IncomingLightning {} },
107+
time: SystemTime::now()
108+
.duration_since(SystemTime::UNIX_EPOCH)
109+
.unwrap(),
110+
},
111+
)
112+
.await;
111113
}
112114
}
113115

@@ -141,7 +143,7 @@ impl RebalanceTrigger for OrangeTrigger {
141143

142144
fn needs_onchain_rebalance(&self) -> impl Future<Output = Option<TriggerParams>> + Send {
143145
async move {
144-
let rebalance_enabled = store::get_rebalance_enabled(self.store.as_ref());
146+
let rebalance_enabled = store::get_rebalance_enabled(self.store.as_ref()).await;
145147
if !rebalance_enabled {
146148
return None;
147149
}
@@ -215,19 +217,21 @@ impl RebalanceTrigger for OrangeTrigger {
215217
// make sure we have a metadata entry for the triggering transaction
216218
let trigger = PaymentId::SelfCustodial(txid.to_byte_array());
217219
if self.tx_metadata.read().get(&trigger).is_none() {
218-
self.tx_metadata.insert(
219-
trigger,
220-
TxMetadata {
221-
ty: TxType::Payment {
222-
ty: PaymentType::IncomingOnChain {
223-
txid: Some(txid),
220+
self.tx_metadata
221+
.insert(
222+
trigger,
223+
TxMetadata {
224+
ty: TxType::Payment {
225+
ty: PaymentType::IncomingOnChain {
226+
txid: Some(txid),
227+
},
224228
},
229+
time: SystemTime::now()
230+
.duration_since(SystemTime::UNIX_EPOCH)
231+
.unwrap(),
225232
},
226-
time: SystemTime::now()
227-
.duration_since(SystemTime::UNIX_EPOCH)
228-
.unwrap(),
229-
},
230-
);
233+
)
234+
.await;
231235
}
232236

233237
Some(TriggerParams {
@@ -295,7 +299,8 @@ impl graduated_rebalancer::EventHandler for OrangeRebalanceEventHandler {
295299
time: SystemTime::now().duration_since(SystemTime::UNIX_EPOCH).unwrap(),
296300
};
297301
self.tx_metadata
298-
.insert(PaymentId::Trusted(trusted_rebalance_payment_id), metadata);
302+
.insert(PaymentId::Trusted(trusted_rebalance_payment_id), metadata)
303+
.await;
299304
if let Err(e) = self
300305
.event_queue
301306
.add_event(Event::RebalanceInitiated {
@@ -318,6 +323,7 @@ impl graduated_rebalancer::EventHandler for OrangeRebalanceEventHandler {
318323
let triggering_transaction_id = PaymentId::Trusted(trigger_id);
319324
self.tx_metadata
320325
.set_tx_caused_rebalance(&triggering_transaction_id)
326+
.await
321327
.expect("Failed to write metadata for rebalance transaction");
322328
let metadata = TxMetadata {
323329
ty: TxType::TrustedToLightning {
@@ -327,8 +333,8 @@ impl graduated_rebalancer::EventHandler for OrangeRebalanceEventHandler {
327333
},
328334
time: SystemTime::now().duration_since(SystemTime::UNIX_EPOCH).unwrap(),
329335
};
330-
self.tx_metadata.upsert(PaymentId::Trusted(rebalance_id), metadata);
331-
self.tx_metadata.insert(PaymentId::SelfCustodial(lightning_id), metadata);
336+
self.tx_metadata.upsert(PaymentId::Trusted(rebalance_id), metadata).await;
337+
self.tx_metadata.insert(PaymentId::SelfCustodial(lightning_id), metadata).await;
332338

333339
let event_queue = Arc::clone(&self.event_queue);
334340
let logger = Arc::clone(&self.logger);
@@ -357,13 +363,15 @@ impl graduated_rebalancer::EventHandler for OrangeRebalanceEventHandler {
357363
let trigger_id = PaymentId::SelfCustodial(triggering_txid.to_byte_array());
358364
self.tx_metadata
359365
.set_tx_caused_rebalance(&trigger_id)
366+
.await
360367
.expect("Failed to write metadata for onchain rebalance transaction");
361368
let metadata = TxMetadata {
362369
ty: TxType::OnchainToLightning { channel_txid: chan_txid, triggering_txid },
363370
time: SystemTime::now().duration_since(SystemTime::UNIX_EPOCH).unwrap(),
364371
};
365372
self.tx_metadata
366-
.insert(PaymentId::SelfCustodial(chan_txid.to_byte_array()), metadata);
373+
.insert(PaymentId::SelfCustodial(chan_txid.to_byte_array()), metadata)
374+
.await;
367375
},
368376
}
369377
})

0 commit comments

Comments
 (0)