Skip to content

Commit 1329914

Browse files
joostjagerjoostjager
committed
Async kvstore
1 parent 3d381e5 commit 1329914

6 files changed

Lines changed: 99 additions & 94 deletions

File tree

‎Cargo.toml‎

Lines changed: 12 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -52,17 +52,17 @@ default = []
5252
#lightning-liquidity = { git = "https://github.com/lightningdevkit/rust-lightning", branch = "main" }
5353
#lightning-macros = { git = "https://github.com/lightningdevkit/rust-lightning", branch = "main" }
5454

55-
lightning = { git = "https://github.com/joostjager/rust-lightning", rev = "03032ddf311d65bc6e75deff64e94d62f7a2f73a", features = ["std"] }
56-
lightning-types = { git = "https://github.com/joostjager/rust-lightning", rev = "03032ddf311d65bc6e75deff64e94d62f7a2f73a" }
57-
lightning-invoice = { git = "https://github.com/joostjager/rust-lightning", rev = "03032ddf311d65bc6e75deff64e94d62f7a2f73a", features = ["std"] }
58-
lightning-net-tokio = { git = "https://github.com/joostjager/rust-lightning", rev = "03032ddf311d65bc6e75deff64e94d62f7a2f73a" }
59-
lightning-persister = { git = "https://github.com/joostjager/rust-lightning", rev = "03032ddf311d65bc6e75deff64e94d62f7a2f73a" }
60-
lightning-background-processor = { git = "https://github.com/joostjager/rust-lightning", rev = "03032ddf311d65bc6e75deff64e94d62f7a2f73a" }
61-
lightning-rapid-gossip-sync = { git = "https://github.com/joostjager/rust-lightning", rev = "03032ddf311d65bc6e75deff64e94d62f7a2f73a" }
62-
lightning-block-sync = { git = "https://github.com/joostjager/rust-lightning", rev = "03032ddf311d65bc6e75deff64e94d62f7a2f73a", features = ["rest-client", "rpc-client", "tokio"] }
63-
lightning-transaction-sync = { git = "https://github.com/joostjager/rust-lightning", rev = "03032ddf311d65bc6e75deff64e94d62f7a2f73a", features = ["esplora-async-https", "electrum", "time"] }
64-
lightning-liquidity = { git = "https://github.com/joostjager/rust-lightning", rev = "03032ddf311d65bc6e75deff64e94d62f7a2f73a" }
65-
lightning-macros = { git = "https://github.com/joostjager/rust-lightning", rev = "03032ddf311d65bc6e75deff64e94d62f7a2f73a" }
55+
lightning = { git = "https://github.com/joostjager/rust-lightning", branch = "async-persister", features = ["std"] }
56+
lightning-types = { git = "https://github.com/joostjager/rust-lightning", branch = "async-persister" }
57+
lightning-invoice = { git = "https://github.com/joostjager/rust-lightning", branch = "async-persister", features = ["std"] }
58+
lightning-net-tokio = { git = "https://github.com/joostjager/rust-lightning", branch = "async-persister" }
59+
lightning-persister = { git = "https://github.com/joostjager/rust-lightning", branch = "async-persister" }
60+
lightning-background-processor = { git = "https://github.com/joostjager/rust-lightning", branch = "async-persister" }
61+
lightning-rapid-gossip-sync = { git = "https://github.com/joostjager/rust-lightning", branch = "async-persister" }
62+
lightning-block-sync = { git = "https://github.com/joostjager/rust-lightning", branch = "async-persister", features = ["rest-client", "rpc-client", "tokio"] }
63+
lightning-transaction-sync = { git = "https://github.com/joostjager/rust-lightning", branch = "async-persister", features = ["esplora-async-https", "electrum", "time"] }
64+
lightning-liquidity = { git = "https://github.com/joostjager/rust-lightning", branch = "async-persister" }
65+
lightning-macros = { git = "https://github.com/joostjager/rust-lightning", branch = "async-persister" }
6666

6767
#lightning = { path = "../rust-lightning/lightning", features = ["std"] }
6868
#lightning-types = { path = "../rust-lightning/lightning-types" }
@@ -108,7 +108,7 @@ winapi = { version = "0.3", features = ["winbase"] }
108108
[dev-dependencies]
109109
#lightning = { version = "0.1.0", features = ["std", "_test_utils"] }
110110
#lightning = { git = "https://github.com/lightningdevkit/rust-lightning", branch="main", features = ["std", "_test_utils"] }
111-
lightning = { git = "https://github.com/joostjager/rust-lightning", rev = "03032ddf311d65bc6e75deff64e94d62f7a2f73a", features = ["std", "_test_utils"] }
111+
lightning = { git = "https://github.com/joostjager/rust-lightning", branch = "async-persister", features = ["std", "_test_utils"] }
112112
#lightning = { path = "../rust-lightning/lightning", features = ["std", "_test_utils"] }
113113
proptest = "1.0.0"
114114
regex = "1.5.6"

‎src/builder.rs‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -34,7 +34,7 @@ use crate::types::{
3434
OnionMessenger, PaymentStore, PeerManager,
3535
};
3636
use crate::wallet::persist::KVStoreWalletPersister;
37-
use crate::wallet::Wallet;
37+
use crate::wallet::{Wallet, WalletKeysManager};
3838
use crate::{Node, NodeMetrics};
3939

4040
use lightning::chain::{chainmonitor, BestBlock, Watch};
@@ -1532,7 +1532,7 @@ fn build_with_store_internal(
15321532
Ok(output_sweeper) => Arc::new(output_sweeper),
15331533
Err(e) => {
15341534
if e.kind() == std::io::ErrorKind::NotFound {
1535-
Arc::new(OutputSweeper::new(
1535+
Arc::new(OutputSweeper::new_with_kv_store_sync(
15361536
channel_manager.current_best_block(),
15371537
Arc::clone(&tx_broadcaster),
15381538
Arc::clone(&fee_estimator),

‎src/event.rs‎

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1062,7 +1062,11 @@ where
10621062
}
10631063
},
10641064
LdkEvent::SpendableOutputs { outputs, channel_id } => {
1065-
match self.output_sweeper.track_spendable_outputs(outputs, channel_id, true, None) {
1065+
match self
1066+
.output_sweeper
1067+
.track_spendable_outputs(outputs, channel_id, true, None)
1068+
.await
1069+
{
10661070
Ok(_) => return Ok(()),
10671071
Err(_) => {
10681072
log_error!(self.logger, "Failed to track spendable outputs");

‎src/io/utils.rs‎

Lines changed: 76 additions & 76 deletions
Original file line numberDiff line numberDiff line change
@@ -253,7 +253,7 @@ pub(crate) fn read_output_sweeper(
253253
kv_store,
254254
logger.clone(),
255255
);
256-
OutputSweeper::read(&mut reader, args).map_err(|e| {
256+
OutputSweeper::read_with_kv_store_sync(&mut reader, args).map_err(|e| {
257257
log_error!(logger, "Failed to deserialize OutputSweeper: {}", e);
258258
std::io::Error::new(std::io::ErrorKind::InvalidData, "Failed to deserialize OutputSweeper")
259259
})
@@ -277,81 +277,81 @@ pub(crate) fn migrate_deprecated_spendable_outputs<L: Deref>(
277277
where
278278
L::Target: LdkLogger,
279279
{
280-
let best_block = sweeper.current_best_block();
281-
282-
for stored_key in kv_store.list(
283-
DEPRECATED_SPENDABLE_OUTPUT_INFO_PERSISTENCE_PRIMARY_NAMESPACE,
284-
DEPRECATED_SPENDABLE_OUTPUT_INFO_PERSISTENCE_SECONDARY_NAMESPACE,
285-
)? {
286-
let mut reader = Cursor::new(kv_store.read(
287-
DEPRECATED_SPENDABLE_OUTPUT_INFO_PERSISTENCE_PRIMARY_NAMESPACE,
288-
DEPRECATED_SPENDABLE_OUTPUT_INFO_PERSISTENCE_SECONDARY_NAMESPACE,
289-
&stored_key,
290-
)?);
291-
let output = DeprecatedSpendableOutputInfo::read(&mut reader).map_err(|e| {
292-
log_error!(logger, "Failed to deserialize SpendableOutputInfo: {}", e);
293-
std::io::Error::new(
294-
std::io::ErrorKind::InvalidData,
295-
"Failed to deserialize SpendableOutputInfo",
296-
)
297-
})?;
298-
let descriptors = vec![output.descriptor.clone()];
299-
let spend_delay = Some(best_block.height + 2);
300-
sweeper
301-
.track_spendable_outputs(descriptors, output.channel_id, true, spend_delay)
302-
.map_err(|_| {
303-
log_error!(logger, "Failed to track spendable outputs. Aborting migration, will retry in the future.");
304-
std::io::Error::new(
305-
std::io::ErrorKind::InvalidData,
306-
"Failed to track spendable outputs. Aborting migration, will retry in the future.",
307-
)
308-
})?;
309-
310-
if let Some(tracked_spendable_output) =
311-
sweeper.tracked_spendable_outputs().iter().find(|o| o.descriptor == output.descriptor)
312-
{
313-
match tracked_spendable_output.status {
314-
OutputSpendStatus::PendingInitialBroadcast { delayed_until_height } => {
315-
if delayed_until_height == spend_delay {
316-
kv_store.remove(
317-
DEPRECATED_SPENDABLE_OUTPUT_INFO_PERSISTENCE_PRIMARY_NAMESPACE,
318-
DEPRECATED_SPENDABLE_OUTPUT_INFO_PERSISTENCE_SECONDARY_NAMESPACE,
319-
&stored_key,
320-
false,
321-
)?;
322-
} else {
323-
debug_assert!(false, "Unexpected status in OutputSweeper migration.");
324-
log_error!(logger, "Unexpected status in OutputSweeper migration.");
325-
return Err(std::io::Error::new(
326-
std::io::ErrorKind::Other,
327-
"Failed to migrate OutputSweeper state.",
328-
));
329-
}
330-
},
331-
_ => {
332-
debug_assert!(false, "Unexpected status in OutputSweeper migration.");
333-
log_error!(logger, "Unexpected status in OutputSweeper migration.");
334-
return Err(std::io::Error::new(
335-
std::io::ErrorKind::Other,
336-
"Failed to migrate OutputSweeper state.",
337-
));
338-
},
339-
}
340-
} else {
341-
debug_assert!(
342-
false,
343-
"OutputSweeper failed to track and persist outputs during migration."
344-
);
345-
log_error!(
346-
logger,
347-
"OutputSweeper failed to track and persist outputs during migration."
348-
);
349-
return Err(std::io::Error::new(
350-
std::io::ErrorKind::Other,
351-
"Failed to migrate OutputSweeper state.",
352-
));
353-
}
354-
}
280+
// let best_block = sweeper.current_best_block();
281+
282+
// for stored_key in kv_store.list(
283+
// DEPRECATED_SPENDABLE_OUTPUT_INFO_PERSISTENCE_PRIMARY_NAMESPACE,
284+
// DEPRECATED_SPENDABLE_OUTPUT_INFO_PERSISTENCE_SECONDARY_NAMESPACE,
285+
// )? {
286+
// let mut reader = Cursor::new(kv_store.read(
287+
// DEPRECATED_SPENDABLE_OUTPUT_INFO_PERSISTENCE_PRIMARY_NAMESPACE,
288+
// DEPRECATED_SPENDABLE_OUTPUT_INFO_PERSISTENCE_SECONDARY_NAMESPACE,
289+
// &stored_key,
290+
// )?);
291+
// let output = DeprecatedSpendableOutputInfo::read(&mut reader).map_err(|e| {
292+
// log_error!(logger, "Failed to deserialize SpendableOutputInfo: {}", e);
293+
// std::io::Error::new(
294+
// std::io::ErrorKind::InvalidData,
295+
// "Failed to deserialize SpendableOutputInfo",
296+
// )
297+
// })?;
298+
// let descriptors = vec![output.descriptor.clone()];
299+
// let spend_delay = Some(best_block.height + 2);
300+
// sweeper
301+
// .track_spendable_outputs(descriptors, output.channel_id, true, spend_delay)
302+
// .map_err(|_| {
303+
// log_error!(logger, "Failed to track spendable outputs. Aborting migration, will retry in the future.");
304+
// std::io::Error::new(
305+
// std::io::ErrorKind::InvalidData,
306+
// "Failed to track spendable outputs. Aborting migration, will retry in the future.",
307+
// )
308+
// })?;
309+
310+
// if let Some(tracked_spendable_output) =
311+
// sweeper.tracked_spendable_outputs().iter().find(|o| o.descriptor == output.descriptor)
312+
// {
313+
// match tracked_spendable_output.status {
314+
// OutputSpendStatus::PendingInitialBroadcast { delayed_until_height } => {
315+
// if delayed_until_height == spend_delay {
316+
// kv_store.remove(
317+
// DEPRECATED_SPENDABLE_OUTPUT_INFO_PERSISTENCE_PRIMARY_NAMESPACE,
318+
// DEPRECATED_SPENDABLE_OUTPUT_INFO_PERSISTENCE_SECONDARY_NAMESPACE,
319+
// &stored_key,
320+
// false,
321+
// )?;
322+
// } else {
323+
// debug_assert!(false, "Unexpected status in OutputSweeper migration.");
324+
// log_error!(logger, "Unexpected status in OutputSweeper migration.");
325+
// return Err(std::io::Error::new(
326+
// std::io::ErrorKind::Other,
327+
// "Failed to migrate OutputSweeper state.",
328+
// ));
329+
// }
330+
// },
331+
// _ => {
332+
// debug_assert!(false, "Unexpected status in OutputSweeper migration.");
333+
// log_error!(logger, "Unexpected status in OutputSweeper migration.");
334+
// return Err(std::io::Error::new(
335+
// std::io::ErrorKind::Other,
336+
// "Failed to migrate OutputSweeper state.",
337+
// ));
338+
// },
339+
// }
340+
// } else {
341+
// debug_assert!(
342+
// false,
343+
// "OutputSweeper failed to track and persist outputs during migration."
344+
// );
345+
// log_error!(
346+
// logger,
347+
// "OutputSweeper failed to track and persist outputs during migration."
348+
// );
349+
// return Err(std::io::Error::new(
350+
// std::io::ErrorKind::Other,
351+
// "Failed to migrate OutputSweeper state.",
352+
// ));
353+
// }
354+
// }
355355

356356
Ok(())
357357
}

‎src/liquidity.rs‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -483,7 +483,7 @@ where
483483
if token != Some(required) {
484484
log_error!(
485485
self.logger,
486-
"Rejecting LSPS2 request {:?} from counterparty {} as the client provided an invalid token.",
486+
"Rejecting LSPS2 request {:?} from counterparty {} as the client provided an invalid token.",
487487
request_id,
488488
counterparty_node_id
489489
);

‎src/types.rs‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -26,9 +26,10 @@ use lightning::routing::router::DefaultRouter;
2626
use lightning::routing::scoring::{ProbabilisticScorer, ProbabilisticScoringFeeParameters};
2727
use lightning::sign::InMemorySigner;
2828
use lightning::util::persist::KVStoreSync;
29+
use lightning::util::persist::KVStoreSyncWrapper;
2930
use lightning::util::ser::{Readable, Writeable, Writer};
30-
use lightning::util::sweep::OutputSweeper;
3131

32+
use lightning::util::sweep::OutputSweeper;
3233
use lightning_block_sync::gossip::{GossipVerifier, UtxoSource};
3334

3435
use lightning_net_tokio::SocketDescriptor;
@@ -131,7 +132,7 @@ pub(crate) type Sweeper = OutputSweeper<
131132
Arc<KeysManager>,
132133
Arc<OnchainFeeEstimator>,
133134
Arc<ChainSource>,
134-
Arc<DynStore>,
135+
KVStoreSyncWrapper<Arc<DynStore>>,
135136
Arc<Logger>,
136137
Arc<KeysManager>,
137138
>;

0 commit comments

Comments
 (0)