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.

Author: tnull

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};

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 {

src/io/utils.rs

@@ -281,22 +281,59 @@ where
 {
 	let mut res = Vec::new();
 
-	for stored_key in KVStore::list(
+	let mut stored_keys = KVStore::list(
 		&*kv_store,
 		PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE,
 		PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE,
 	)
-	.await?
-	{
-		let mut reader = Cursor::new(
-			KVStore::read(
+	.await?;
+
+	const BATCH_SIZE: usize = 50;
+
+	let mut set = tokio::task::JoinSet::new();
+
+	// Fill JoinSet with tasks if possible
+	while set.len() < BATCH_SIZE && !stored_keys.is_empty() {
+		if let Some(next_key) = stored_keys.pop() {
+			let fut = KVStore::read(
 				&*kv_store,
 				PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE,
 				PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE,
-				&stored_key,
-			)
-			.await?,
-		);
+				&next_key,
+			);
+			set.spawn(fut);
+			debug_assert!(set.len() <= BATCH_SIZE);
+		}
+	}
+
+	while let Some(read_res) = set.join_next().await {
+		// Exit early if we get an IO error.
+		let read_res = read_res
+			.map_err(|e| {
+				log_error!(logger, "Failed to read PaymentDetails: {}", e);
+				set.abort_all();
+				e
+			})?
+			.map_err(|e| {
+				log_error!(logger, "Failed to read PaymentDetails: {}", e);
+				set.abort_all();
+				e
+			})?;
+
+		// Refill set for every finished future, if we still have something to do.
+		if let Some(next_key) = stored_keys.pop() {
+			let fut = KVStore::read(
+				&*kv_store,
+				PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE,
+				PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE,
+				&next_key,
+			);
+			set.spawn(fut);
+			debug_assert!(set.len() <= BATCH_SIZE);
+		}
+
+		// Handle result.
+		let mut reader = Cursor::new(read_res);
 		let payment = PaymentDetails::read(&mut reader).map_err(|e| {
 			log_error!(logger, "Failed to deserialize PaymentDetails: {}", e);
 			std::io::Error::new(
@@ -306,6 +343,10 @@ where
 		})?;
 		res.push(payment);
 	}
+
+	debug_assert!(set.is_empty());
+	debug_assert!(stored_keys.is_empty());
+
 	Ok(res)
 }
 

src/payment/unified_qr.rs

@@ -18,6 +18,8 @@ use bip21::de::ParamKind;
 use bip21::{DeserializationError, DeserializeParams, Param, SerializeParams};
 use bitcoin::address::{NetworkChecked, NetworkUnchecked};
 use bitcoin::{Amount, Txid};
+use bitcoin_payment_instructions::amount::Amount as BPIAmount;
+use bitcoin_payment_instructions::{PaymentInstructions, PaymentMethod};
 use lightning::ln::channelmanager::PaymentId;
 use lightning::offers::offer::Offer;
 use lightning::routing::router::RouteParametersConfig;
@@ -310,6 +312,7 @@ impl DeserializationError for Extras {
 mod tests {
 	use std::str::FromStr;
 
+	use bitcoin::address::NetworkUnchecked;
 	use bitcoin::{Address, Network};
 
 	use super::*;

src/types.rs

@@ -9,7 +9,8 @@ use std::fmt;
 use std::sync::{Arc, Mutex};
 
 use bitcoin::secp256k1::PublicKey;
-use bitcoin::OutPoint;
+use bitcoin::{OutPoint, ScriptBuf};
+use bitcoin_payment_instructions::onion_message_resolver::LDKOnionMessageDNSSECHrnResolver;
 use lightning::chain::chainmonitor;
 use lightning::impl_writeable_tlv_based;
 use lightning::ln::channel_state::ChannelDetails as LdkChannelDetails;
@@ -27,6 +28,7 @@ use lightning_block_sync::gossip::{GossipVerifier, UtxoSource};
 use lightning_liquidity::utils::time::DefaultTimeProvider;
 use lightning_net_tokio::SocketDescriptor;
 
+use crate::chain::bitcoind::UtxoSourceClient;
 use crate::chain::ChainSource;
 use crate::config::ChannelConfig;
 use crate::data_store::DataStore;