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/io/utils.rs

@@ -236,22 +236,49 @@ 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 = 20;
+
+	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??;
+
+		// 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(
@@ -261,6 +288,10 @@ where
 		})?;
 		res.push(payment);
 	}
+
+	debug_assert!(set.is_empty());
+	debug_assert!(stored_keys.is_empty());
+
 	Ok(res)
 }
 

src/runtime.rs

@@ -207,7 +207,7 @@ impl Runtime {
 		);
 	}
 
-	fn handle(&self) -> &tokio::runtime::Handle {
+	pub fn handle(&self) -> &tokio::runtime::Handle {
 		match &self.mode {
 			RuntimeMode::Owned(rt) => rt.handle(),
 			RuntimeMode::Handle(handle) => handle,