Switch to the new highly-parallel ChannelMonitor reader

Upstream LDK added the ability to read `ChannelMonitor`s from
storage in parallel, which we switch to here.

Author: TheBlueMatt

src/builder.rs

@@ -51,7 +51,7 @@ use crate::connection::ConnectionManager;
 use crate::entropy::NodeEntropy;
 use crate::event::EventQueue;
 use crate::fee_estimator::OnchainFeeEstimator;
-use crate::gossip::GossipSource;
+use crate::gossip::{GossipSource, RuntimeSpawner};
 use crate::io::sqlite_store::SqliteStore;
 use crate::io::utils::{
 	read_event_queue, read_external_pathfinding_scores_from_cache, read_network_graph,
@@ -72,8 +72,9 @@ use crate::peer_store::PeerStore;
 use crate::runtime::Runtime;
 use crate::tx_broadcaster::TransactionBroadcaster;
 use crate::types::{
-	ChainMonitor, ChannelManager, DynStore, DynStoreWrapper, GossipSync, Graph, KeysManager,
-	MessageRouter, OnionMessenger, PaymentStore, PeerManager, Persister, SyncAndAsyncKVStore,
+	AsyncPersister, ChainMonitor, ChannelManager, DynStore, DynStoreWrapper, GossipSync, Graph,
+	KeysManager, MessageRouter, OnionMessenger, PaymentStore, PeerManager, Persister,
+	SyncAndAsyncKVStore,
 };
 use crate::wallet::persist::KVStoreWalletPersister;
 use crate::wallet::Wallet;
@@ -1261,8 +1262,9 @@ fn build_with_store_internal(
 	));
 
 	let peer_storage_key = keys_manager.get_peer_storage_key();
-	let persister = Arc::new(Persister::new(
+	let monitor_reader = Arc::new(AsyncPersister::new(
 		Arc::clone(&kv_store),
+		RuntimeSpawner::new(Arc::clone(&runtime)),
 		Arc::clone(&logger),
 		PERSISTER_MAX_PENDING_UPDATES,
 		Arc::clone(&keys_manager),
@@ -1272,7 +1274,9 @@ fn build_with_store_internal(
 	));
 
 	// Read ChannelMonitor state from store
-	let channel_monitors = match persister.read_all_channel_monitors_with_updates() {
+	let monitor_read_result =
+		runtime.block_on(monitor_reader.read_all_channel_monitors_with_updates_parallel());
+	let channel_monitors = match monitor_read_result {
 		Ok(monitors) => monitors,
 		Err(e) => {
 			if e.kind() == lightning::io::ErrorKind::NotFound {
@@ -1284,6 +1288,16 @@ fn build_with_store_internal(
 		},
 	};
 
+	let persister = Arc::new(Persister::new(
+		Arc::clone(&kv_store),
+		Arc::clone(&logger),
+		PERSISTER_MAX_PENDING_UPDATES,
+		Arc::clone(&keys_manager),
+		Arc::clone(&keys_manager),
+		Arc::clone(&tx_broadcaster),
+		Arc::clone(&fee_estimator),
+	));
+
 	// Initialize the ChainMonitor
 	let chain_monitor: Arc<ChainMonitor> = Arc::new(chainmonitor::ChainMonitor::new(
 		Some(Arc::clone(&chain_source)),

src/types.rs

@@ -23,7 +23,9 @@ use lightning::routing::gossip;
 use lightning::routing::router::DefaultRouter;
 use lightning::routing::scoring::{CombinedScorer, ProbabilisticScoringFeeParameters};
 use lightning::sign::InMemorySigner;
-use lightning::util::persist::{KVStore, KVStoreSync, MonitorUpdatingPersister};
+use lightning::util::persist::{
+	KVStore, KVStoreSync, MonitorUpdatingPersister, MonitorUpdatingPersisterAsync,
+};
 use lightning::util::ser::{Readable, Writeable, Writer};
 use lightning::util::sweep::OutputSweeper;
 use lightning_block_sync::gossip::GossipVerifier;
@@ -185,6 +187,16 @@ impl<T: SyncAndAsyncKVStore + Send + Sync> DynStoreTrait for DynStoreWrapper<T>
 	}
 }
 
+pub(crate) type AsyncPersister = MonitorUpdatingPersisterAsync<
+	Arc<DynStore>,
+	RuntimeSpawner,
+	Arc<Logger>,
+	Arc<KeysManager>,
+	Arc<KeysManager>,
+	Arc<Broadcaster>,
+	Arc<OnchainFeeEstimator>,
+>;
+
 pub type Persister = MonitorUpdatingPersister<
 	Arc<DynStore>,
 	Arc<Logger>,