Parallelize store reads in init

Since I was editing the init logic anyway I couldn't resist going
ahead and parallelizing various read calls. Since we added support
for an async `KVStore` in LDK 0.2/ldk-node 0.7, we can now
practically do initialization reads in parallel. Thus, rather than
making a long series of read calls in `build`, we use `tokio::join`
to reduce the number of round-trips to our backing store, which
should be a very large win for initialization cost on those using
remote storage (e.g. VSS).

Sadly we can't trivially do all our reads in one go, we need the
payment history to initialize the BDK wallet, which is used in the
`Walet` object which is referenced in our `KeysManager`. Thus we
first read the payment store and node metrics before moving on.
Then, we need a reference to the `NetworkGraph` when we build the
scorer. While we could/eventually should move to reading the
*bytes* for the scorer while reading the graph and only building
the scorer later, that's a larger refactor we leave for later.

In the end, we end up with:
 * 1 round-trip to load the payment history and node metrics,
 * 2 round-trips to load ChannelMonitors and NetworkGraph (where
   there's an internal extra round-trip after listing the monitor
   updates for a monitor),
 * 1 round-trip to validate bitcoind RPC/REST access for those
   using bitcoind as a chain source,
 * 1 round-trip to load various smaller LDK and ldk-node objects,
 * and 1 additional round-trip to drop the rgs snapshot timestamp
   for nodes using P2P network gossip syncing

for a total of 4 round-trips in the common case and 6 for nodes
using less common chain source and gossip sync sources.

We then have additional round-trips to our storage and chain source
during node start, but those are in many cases already async.

Author: TheBlueMatt

src/builder.rs

@@ -33,7 +33,7 @@ use lightning::routing::scoring::{
 };
 use lightning::sign::{EntropySource, NodeSigner};
 use lightning::util::persist::{
-	KVStoreSync, CHANNEL_MANAGER_PERSISTENCE_KEY, CHANNEL_MANAGER_PERSISTENCE_PRIMARY_NAMESPACE,
+	KVStore, CHANNEL_MANAGER_PERSISTENCE_KEY, CHANNEL_MANAGER_PERSISTENCE_PRIMARY_NAMESPACE,
 	CHANNEL_MANAGER_PERSISTENCE_SECONDARY_NAMESPACE,
 };
 use lightning::util::ser::ReadableArgs;
@@ -1052,10 +1052,20 @@ fn build_with_store_internal(
 		}
 	}
 
+	let tx_broadcaster = Arc::new(TransactionBroadcaster::new(Arc::clone(&logger)));
+	let fee_estimator = Arc::new(OnchainFeeEstimator::new());
+
+	let kv_store_ref = Arc::clone(&kv_store);
+	let logger_ref = Arc::clone(&logger);
+	let (payment_store_res, node_metris_res) = runtime.block_on(async move {
+		tokio::join!(
+			read_payments(&*kv_store_ref, Arc::clone(&logger_ref)),
+			read_node_metrics(&*kv_store_ref, Arc::clone(&logger_ref)),
+		)
+	});
+
 	// Initialize the status fields.
-	let node_metrics = match runtime
-		.block_on(async { read_node_metrics(&*kv_store, Arc::clone(&logger)).await })
-	{
+	let node_metrics = match node_metris_res {
 		Ok(metrics) => Arc::new(RwLock::new(metrics)),
 		Err(e) => {
 			if e.kind() == std::io::ErrorKind::NotFound {
@@ -1066,23 +1076,20 @@ 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 runtime.block_on(async { read_payments(&*kv_store, Arc::clone(&logger)).await }) {
-			Ok(payments) => Arc::new(PaymentStore::new(
-				payments,
-				PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE.to_string(),
-				PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE.to_string(),
-				Arc::clone(&kv_store),
-				Arc::clone(&logger),
-			)),
-			Err(e) => {
-				log_error!(logger, "Failed to read payment data from store: {}", e);
-				return Err(BuildError::ReadFailed);
-			},
-		};
+	let payment_store = match payment_store_res {
+		Ok(payments) => Arc::new(PaymentStore::new(
+			payments,
+			PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE.to_string(),
+			PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE.to_string(),
+			Arc::clone(&kv_store),
+			Arc::clone(&logger),
+		)),
+		Err(e) => {
+			log_error!(logger, "Failed to read payment data from store: {}", e);
+			return Err(BuildError::ReadFailed);
+		},
+	};
 
 	let (chain_source, chain_tip_opt) = match chain_data_source_config {
 		Some(ChainDataSourceConfig::Esplora { server_url, headers, sync_config }) => {
@@ -1273,10 +1280,18 @@ fn build_with_store_internal(
 		Arc::clone(&fee_estimator),
 	));
 
+	// Read ChannelMonitors and the NetworkGraph
+	let kv_store_ref = Arc::clone(&kv_store);
+	let logger_ref = Arc::clone(&logger);
+	let (monitor_read_res, network_graph_res) = runtime.block_on(async move {
+		tokio::join!(
+			monitor_reader.read_all_channel_monitors_with_updates_parallel(),
+			read_network_graph(&*kv_store_ref, logger_ref),
+		)
+	});
+
 	// Read ChannelMonitor state from store
-	let monitor_read_result =
-		runtime.block_on(monitor_reader.read_all_channel_monitors_with_updates_parallel());
-	let channel_monitors = match monitor_read_result {
+	let channel_monitors = match monitor_read_res {
 		Ok(monitors) => monitors,
 		Err(e) => {
 			if e.kind() == lightning::io::ErrorKind::NotFound {
@@ -1310,9 +1325,7 @@ fn build_with_store_internal(
 	));
 
 	// Initialize the network graph, scorer, and router
-	let network_graph = match runtime
-		.block_on(async { read_network_graph(&*kv_store, Arc::clone(&logger)).await })
-	{
+	let network_graph = match network_graph_res {
 		Ok(graph) => Arc::new(graph),
 		Err(e) => {
 			if e.kind() == std::io::ErrorKind::NotFound {
@@ -1324,9 +1337,42 @@ fn build_with_store_internal(
 		},
 	};
 
-	let local_scorer = match runtime.block_on(async {
-		read_scorer(&*kv_store, Arc::clone(&network_graph), Arc::clone(&logger)).await
-	}) {
+	// Read various smaller LDK and ldk-node objects from the store
+	let kv_store_ref = Arc::clone(&kv_store);
+	let logger_ref = Arc::clone(&logger);
+	let network_graph_ref = Arc::clone(&network_graph);
+	let output_sweeper_future = read_output_sweeper(
+		Arc::clone(&tx_broadcaster),
+		Arc::clone(&fee_estimator),
+		Arc::clone(&chain_source),
+		Arc::clone(&keys_manager),
+		Arc::clone(&kv_store_ref),
+		Arc::clone(&logger_ref),
+	);
+	let (
+		scorer_res,
+		external_scores_res,
+		channel_manager_bytes_res,
+		sweeper_bytes_res,
+		event_queue_res,
+		peer_info_res,
+	) = runtime.block_on(async move {
+		tokio::join!(
+			read_scorer(&*kv_store_ref, network_graph_ref, Arc::clone(&logger_ref)),
+			read_external_pathfinding_scores_from_cache(&*kv_store_ref, Arc::clone(&logger_ref)),
+			KVStore::read(
+				&*kv_store_ref,
+				CHANNEL_MANAGER_PERSISTENCE_PRIMARY_NAMESPACE,
+				CHANNEL_MANAGER_PERSISTENCE_SECONDARY_NAMESPACE,
+				CHANNEL_MANAGER_PERSISTENCE_KEY,
+			),
+			output_sweeper_future,
+			read_event_queue(Arc::clone(&kv_store_ref), Arc::clone(&logger_ref)),
+			read_peer_info(Arc::clone(&kv_store_ref), Arc::clone(&logger_ref)),
+		)
+	});
+
+	let local_scorer = match scorer_res {
 		Ok(scorer) => scorer,
 		Err(e) => {
 			if e.kind() == std::io::ErrorKind::NotFound {
@@ -1342,9 +1388,7 @@ fn build_with_store_internal(
 	let scorer = Arc::new(Mutex::new(CombinedScorer::new(local_scorer)));
 
 	// Restore external pathfinding scores from cache if possible.
-	match runtime.block_on(async {
-		read_external_pathfinding_scores_from_cache(&*kv_store, Arc::clone(&logger)).await
-	}) {
+	match external_scores_res {
 		Ok(external_scores) => {
 			scorer.lock().unwrap().merge(external_scores, cur_time);
 			log_trace!(logger, "External scores from cache merged successfully");
@@ -1397,12 +1441,7 @@ fn build_with_store_internal(
 
 	// Initialize the ChannelManager
 	let channel_manager = {
-		if let Ok(reader) = KVStoreSync::read(
-			&*kv_store,
-			CHANNEL_MANAGER_PERSISTENCE_PRIMARY_NAMESPACE,
-			CHANNEL_MANAGER_PERSISTENCE_SECONDARY_NAMESPACE,
-			CHANNEL_MANAGER_PERSISTENCE_KEY,
-		) {
+		if let Ok(reader) = channel_manager_bytes_res {
 			let channel_monitor_references =
 				channel_monitors.iter().map(|(_, chanmon)| chanmon).collect();
 			let read_args = ChannelManagerReadArgs::new(
@@ -1627,17 +1666,7 @@ fn build_with_store_internal(
 	let connection_manager =
 		Arc::new(ConnectionManager::new(Arc::clone(&peer_manager), 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
-	}) {
+	let output_sweeper = match sweeper_bytes_res {
 		Ok(output_sweeper) => Arc::new(output_sweeper),
 		Err(e) => {
 			if e.kind() == std::io::ErrorKind::NotFound {
@@ -1658,9 +1687,7 @@ fn build_with_store_internal(
 		},
 	};
 
-	let event_queue = match runtime
-		.block_on(async { read_event_queue(Arc::clone(&kv_store), Arc::clone(&logger)).await })
-	{
+	let event_queue = match event_queue_res {
 		Ok(event_queue) => Arc::new(event_queue),
 		Err(e) => {
 			if e.kind() == std::io::ErrorKind::NotFound {
@@ -1672,9 +1699,7 @@ fn build_with_store_internal(
 		},
 	};
 
-	let peer_store = match runtime
-		.block_on(async { read_peer_info(Arc::clone(&kv_store), Arc::clone(&logger)).await })
-	{
+	let peer_store = match peer_info_res {
 		Ok(peer_store) => Arc::new(peer_store),
 		Err(e) => {
 			if e.kind() == std::io::ErrorKind::NotFound {