Parallelize read_payments (#7)

* Use async `KVStore` for `read_X` util methods

Rather than using `KVStoreSync` we now use the async `KVStore`
implementation for most `read_X` util methods used during node building.

This is a first step towards making node building/startup entirely async
eventually.

* Parallelize `read_payments`

Previously, we would read entries of our payment store sequentially.
This is more or less fine when we read from a local store, but when we
read from a remote (e.g., VSS) store, all the latency could result in
considerable slowdown during startup. Here, we opt to read store entries
in batches.

* Add test for payment persistence after node restart

Add integration test that verifies 200 payments are correctly
persisted and retrievable via `list_payments` after restarting
a node.

Co-Authored-By: Claude AI

---------

Co-authored-by: Elias Rohrer <[email protected]>

Author: amackillop

Cargo.toml

@@ -92,6 +92,8 @@ log = { version = "0.4.22", default-features = false, features = ["std"]}
 
 vss-client = { package = "vss-client-ng", version = "0.4" }
 prost = { version = "0.11.6", default-features = false}
+#bitcoin-payment-instructions = { version = "0.6" }
+bitcoin-payment-instructions = { git = "https://github.com/tnull/bitcoin-payment-instructions", branch = "2025-12-ldk-node-base" }
 
 [target.'cfg(windows)'.dependencies]
 winapi = { version = "0.3", features = ["winbase"] }

src/builder.rs

@@ -19,7 +19,7 @@ use bip39::Mnemonic;
 use bitcoin::bip32::{ChildNumber, Xpriv};
 use bitcoin::secp256k1::PublicKey;
 use bitcoin::{BlockHash, Network};
-use bdk_chain::{BlockId, TxUpdate};
+use bitcoin_payment_instructions::onion_message_resolver::LDKOnionMessageDNSSECHrnResolver;
 use lightning::chain::{chainmonitor, BestBlock, Watch};
 use lightning::io::Cursor;
 use lightning::ln::channelmanager::{self, ChainParameters, ChannelManagerReadArgs};
@@ -55,7 +55,9 @@ use crate::fee_estimator::OnchainFeeEstimator;
 use crate::gossip::GossipSource;
 use crate::io::sqlite_store::SqliteStore;
 use crate::io::utils::{
-	read_external_pathfinding_scores_from_cache, read_node_metrics, write_node_metrics,
+	read_event_queue, read_external_pathfinding_scores_from_cache, read_network_graph,
+	read_node_metrics, read_output_sweeper, read_payments, read_peer_info, read_scorer,
+	write_node_metrics,
 };
 use crate::io::vss_store::VssStore;
 use crate::io::{
@@ -1233,7 +1235,9 @@ fn build_with_store_internal(
 	}
 
 	// Initialize the status fields.
-	let node_metrics = match read_node_metrics(Arc::clone(&kv_store), Arc::clone(&logger)) {
+	let node_metrics = match runtime
+		.block_on(async { read_node_metrics(Arc::clone(&kv_store), Arc::clone(&logger)).await })
+	{
 		Ok(metrics) => Arc::new(RwLock::new(metrics)),
 		Err(e) => {
 			if e.kind() == std::io::ErrorKind::NotFound {
@@ -1247,7 +1251,9 @@ fn build_with_store_internal(
 	let tx_broadcaster = Arc::new(TransactionBroadcaster::new(Arc::clone(&logger)));
 	let fee_estimator = Arc::new(OnchainFeeEstimator::new());
 
-	let payment_store = match io::utils::read_payments(Arc::clone(&kv_store), Arc::clone(&logger)) {
+	let payment_store = match runtime
+		.block_on(async { read_payments(Arc::clone(&kv_store), Arc::clone(&logger)).await })
+	{
 		Ok(payments) => Arc::new(PaymentStore::new(
 			payments,
 			PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE.to_string(),
@@ -1474,24 +1480,23 @@ fn build_with_store_internal(
 	));
 
 	// Initialize the network graph, scorer, and router
-	let network_graph =
-		match io::utils::read_network_graph(Arc::clone(&kv_store), Arc::clone(&logger)) {
-			Ok(graph) => Arc::new(graph),
-			Err(e) => {
-				if e.kind() == std::io::ErrorKind::NotFound {
-					Arc::new(Graph::new(config.network.into(), Arc::clone(&logger)))
-				} else {
-					log_error!(logger, "Failed to read network graph from store: {}", e);
-					return Err(BuildError::ReadFailed);
-				}
-			},
-		};
+	let network_graph = match runtime
+		.block_on(async { read_network_graph(Arc::clone(&kv_store), Arc::clone(&logger)).await })
+	{
+		Ok(graph) => Arc::new(graph),
+		Err(e) => {
+			if e.kind() == std::io::ErrorKind::NotFound {
+				Arc::new(Graph::new(config.network.into(), Arc::clone(&logger)))
+			} else {
+				log_error!(logger, "Failed to read network graph from store: {}", e);
+				return Err(BuildError::ReadFailed);
+			}
+		},
+	};
 
-	let local_scorer = match io::utils::read_scorer(
-		Arc::clone(&kv_store),
-		Arc::clone(&network_graph),
-		Arc::clone(&logger),
-	) {
+	let local_scorer = match runtime.block_on(async {
+		read_scorer(Arc::clone(&kv_store), Arc::clone(&network_graph), Arc::clone(&logger)).await
+	}) {
 		Ok(scorer) => scorer,
 		Err(e) => {
 			if e.kind() == std::io::ErrorKind::NotFound {
@@ -1507,7 +1512,10 @@ fn build_with_store_internal(
 	let scorer = Arc::new(Mutex::new(CombinedScorer::new(local_scorer)));
 
 	// Restore external pathfinding scores from cache if possible.
-	match read_external_pathfinding_scores_from_cache(Arc::clone(&kv_store), Arc::clone(&logger)) {
+	match runtime.block_on(async {
+		read_external_pathfinding_scores_from_cache(Arc::clone(&kv_store), Arc::clone(&logger))
+			.await
+	}) {
 		Ok(external_scores) => {
 			scorer.lock().unwrap().merge(external_scores, cur_time);
 			log_trace!(logger, "External scores from cache merged successfully");
@@ -1709,7 +1717,8 @@ fn build_with_store_internal(
 		},
 	};
 
-	let event_queue = match io::utils::read_event_queue(Arc::clone(&kv_store), Arc::clone(&logger))
+	let event_queue = match runtime
+		.block_on(async { read_event_queue(Arc::clone(&kv_store), Arc::clone(&logger)).await })
 	{
 		Ok(event_queue) => Arc::new(event_queue),
 		Err(e) => {
@@ -1831,14 +1840,17 @@ fn build_with_store_internal(
 	let connection_manager =
 		Arc::new(ConnectionManager::new(Arc::clone(&peer_manager), Arc::clone(&logger)));
 
-	let output_sweeper = match io::utils::read_output_sweeper(
-		Arc::clone(&tx_broadcaster),
-		Arc::clone(&fee_estimator),
-		Arc::clone(&chain_source),
-		Arc::clone(&keys_manager),
-		Arc::clone(&kv_store),
-		Arc::clone(&logger),
-	) {
+	let output_sweeper = match runtime.block_on(async {
+		read_output_sweeper(
+			Arc::clone(&tx_broadcaster),
+			Arc::clone(&fee_estimator),
+			Arc::clone(&chain_source),
+			Arc::clone(&keys_manager),
+			Arc::clone(&kv_store),
+			Arc::clone(&logger),
+		)
+		.await
+	}) {
 		Ok(output_sweeper) => Arc::new(output_sweeper),
 		Err(e) => {
 			if e.kind() == std::io::ErrorKind::NotFound {
@@ -1859,7 +1871,9 @@ fn build_with_store_internal(
 		},
 	};
 
-	let peer_store = match io::utils::read_peer_info(Arc::clone(&kv_store), Arc::clone(&logger)) {
+	let peer_store = match runtime
+		.block_on(async { read_peer_info(Arc::clone(&kv_store), Arc::clone(&logger)).await })
+	{
 		Ok(peer_store) => Arc::new(peer_store),
 		Err(e) => {
 			if e.kind() == std::io::ErrorKind::NotFound {

src/ffi/types.rs

@@ -29,6 +29,7 @@ use lightning::offers::invoice::Bolt12Invoice as LdkBolt12Invoice;
 pub use lightning::offers::offer::OfferId;
 use lightning::offers::offer::{Amount as LdkAmount, Offer as LdkOffer};
 use lightning::offers::refund::Refund as LdkRefund;
+use lightning::onion_message::dns_resolution::HumanReadableName as LdkHumanReadableName;
 pub use lightning::routing::gossip::{NodeAlias, NodeId, RoutingFees};
 pub use lightning::routing::router::RouteParametersConfig;
 use lightning::util::ser::Writeable;
@@ -54,7 +55,7 @@ pub use crate::logger::{LogLevel, LogRecord, LogWriter};
 pub use crate::payment::store::{
 	ConfirmationStatus, LSPFeeLimits, PaymentDirection, PaymentKind, PaymentStatus,
 };
-pub use crate::payment::QrPaymentResult;
+pub use crate::payment::UnifiedPaymentResult;
 use crate::{hex_utils, SocketAddress, UniffiCustomTypeConverter, UserChannelId};
 
 impl UniffiCustomTypeConverter for PublicKey {