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.

Author: tnull

src/builder.rs

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

@@ -133,38 +133,44 @@ where
 }
 
 /// Read a previously persisted [`NetworkGraph`] from the store.
-pub(crate) fn read_network_graph<L: Deref + Clone>(
+pub(crate) async fn read_network_graph<L: Deref + Clone>(
 	kv_store: Arc<DynStore>, logger: L,
 ) -> Result<NetworkGraph<L>, std::io::Error>
 where
 	L::Target: LdkLogger,
 {
-	let mut reader = Cursor::new(KVStoreSync::read(
-		&*kv_store,
-		NETWORK_GRAPH_PERSISTENCE_PRIMARY_NAMESPACE,
-		NETWORK_GRAPH_PERSISTENCE_SECONDARY_NAMESPACE,
-		NETWORK_GRAPH_PERSISTENCE_KEY,
-	)?);
+	let mut reader = Cursor::new(
+		KVStore::read(
+			&*kv_store,
+			NETWORK_GRAPH_PERSISTENCE_PRIMARY_NAMESPACE,
+			NETWORK_GRAPH_PERSISTENCE_SECONDARY_NAMESPACE,
+			NETWORK_GRAPH_PERSISTENCE_KEY,
+		)
+		.await?,
+	);
 	NetworkGraph::read(&mut reader, logger.clone()).map_err(|e| {
 		log_error!(logger, "Failed to deserialize NetworkGraph: {}", e);
 		std::io::Error::new(std::io::ErrorKind::InvalidData, "Failed to deserialize NetworkGraph")
 	})
 }
 
 /// Read a previously persisted [`ProbabilisticScorer`] from the store.
-pub(crate) fn read_scorer<G: Deref<Target = NetworkGraph<L>>, L: Deref + Clone>(
+pub(crate) async fn read_scorer<G: Deref<Target = NetworkGraph<L>>, L: Deref + Clone>(
 	kv_store: Arc<DynStore>, network_graph: G, logger: L,
 ) -> Result<ProbabilisticScorer<G, L>, std::io::Error>
 where
 	L::Target: LdkLogger,
 {
 	let params = ProbabilisticScoringDecayParameters::default();
-	let mut reader = Cursor::new(KVStoreSync::read(
-		&*kv_store,
-		SCORER_PERSISTENCE_PRIMARY_NAMESPACE,
-		SCORER_PERSISTENCE_SECONDARY_NAMESPACE,
-		SCORER_PERSISTENCE_KEY,
-	)?);
+	let mut reader = Cursor::new(
+		KVStore::read(
+			&*kv_store,
+			SCORER_PERSISTENCE_PRIMARY_NAMESPACE,
+			SCORER_PERSISTENCE_SECONDARY_NAMESPACE,
+			SCORER_PERSISTENCE_KEY,
+		)
+		.await?,
+	);
 	let args = (params, network_graph, logger.clone());
 	ProbabilisticScorer::read(&mut reader, args).map_err(|e| {
 		log_error!(logger, "Failed to deserialize scorer: {}", e);
@@ -173,18 +179,21 @@ where
 }
 
 /// Read previously persisted external pathfinding scores from the cache.
-pub(crate) fn read_external_pathfinding_scores_from_cache<L: Deref>(
+pub(crate) async fn read_external_pathfinding_scores_from_cache<L: Deref>(
 	kv_store: Arc<DynStore>, logger: L,
 ) -> Result<ChannelLiquidities, std::io::Error>
 where
 	L::Target: LdkLogger,
 {
-	let mut reader = Cursor::new(KVStoreSync::read(
-		&*kv_store,
-		SCORER_PERSISTENCE_PRIMARY_NAMESPACE,
-		SCORER_PERSISTENCE_SECONDARY_NAMESPACE,
-		EXTERNAL_PATHFINDING_SCORES_CACHE_KEY,
-	)?);
+	let mut reader = Cursor::new(
+		KVStore::read(
+			&*kv_store,
+			SCORER_PERSISTENCE_PRIMARY_NAMESPACE,
+			SCORER_PERSISTENCE_SECONDARY_NAMESPACE,
+			EXTERNAL_PATHFINDING_SCORES_CACHE_KEY,
+		)
+		.await?,
+	);
 	ChannelLiquidities::read(&mut reader).map_err(|e| {
 		log_error!(logger, "Failed to deserialize scorer: {}", e);
 		std::io::Error::new(std::io::ErrorKind::InvalidData, "Failed to deserialize Scorer")
@@ -220,63 +229,74 @@ where
 }
 
 /// Read previously persisted events from the store.
-pub(crate) fn read_event_queue<L: Deref + Clone>(
+pub(crate) async fn read_event_queue<L: Deref + Clone>(
 	kv_store: Arc<DynStore>, logger: L,
 ) -> Result<EventQueue<L>, std::io::Error>
 where
 	L::Target: LdkLogger,
 {
-	let mut reader = Cursor::new(KVStoreSync::read(
-		&*kv_store,
-		EVENT_QUEUE_PERSISTENCE_PRIMARY_NAMESPACE,
-		EVENT_QUEUE_PERSISTENCE_SECONDARY_NAMESPACE,
-		EVENT_QUEUE_PERSISTENCE_KEY,
-	)?);
+	let mut reader = Cursor::new(
+		KVStore::read(
+			&*kv_store,
+			EVENT_QUEUE_PERSISTENCE_PRIMARY_NAMESPACE,
+			EVENT_QUEUE_PERSISTENCE_SECONDARY_NAMESPACE,
+			EVENT_QUEUE_PERSISTENCE_KEY,
+		)
+		.await?,
+	);
 	EventQueue::read(&mut reader, (kv_store, logger.clone())).map_err(|e| {
 		log_error!(logger, "Failed to deserialize event queue: {}", e);
 		std::io::Error::new(std::io::ErrorKind::InvalidData, "Failed to deserialize EventQueue")
 	})
 }
 
 /// Read previously persisted peer info from the store.
-pub(crate) fn read_peer_info<L: Deref + Clone>(
+pub(crate) async fn read_peer_info<L: Deref + Clone>(
 	kv_store: Arc<DynStore>, logger: L,
 ) -> Result<PeerStore<L>, std::io::Error>
 where
 	L::Target: LdkLogger,
 {
-	let mut reader = Cursor::new(KVStoreSync::read(
-		&*kv_store,
-		PEER_INFO_PERSISTENCE_PRIMARY_NAMESPACE,
-		PEER_INFO_PERSISTENCE_SECONDARY_NAMESPACE,
-		PEER_INFO_PERSISTENCE_KEY,
-	)?);
+	let mut reader = Cursor::new(
+		KVStore::read(
+			&*kv_store,
+			PEER_INFO_PERSISTENCE_PRIMARY_NAMESPACE,
+			PEER_INFO_PERSISTENCE_SECONDARY_NAMESPACE,
+			PEER_INFO_PERSISTENCE_KEY,
+		)
+		.await?,
+	);
 	PeerStore::read(&mut reader, (kv_store, logger.clone())).map_err(|e| {
 		log_error!(logger, "Failed to deserialize peer store: {}", e);
 		std::io::Error::new(std::io::ErrorKind::InvalidData, "Failed to deserialize PeerStore")
 	})
 }
 
 /// Read previously persisted payments information from the store.
-pub(crate) fn read_payments<L: Deref>(
+pub(crate) async fn read_payments<L: Deref>(
 	kv_store: Arc<DynStore>, logger: L,
 ) -> Result<Vec<PaymentDetails>, std::io::Error>
 where
 	L::Target: LdkLogger,
 {
 	let mut res = Vec::new();
 
-	for stored_key in KVStoreSync::list(
+	for stored_key in KVStore::list(
 		&*kv_store,
 		PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE,
 		PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE,
-	)? {
-		let mut reader = Cursor::new(KVStoreSync::read(
-			&*kv_store,
-			PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE,
-			PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE,
-			&stored_key,
-		)?);
+	)
+	.await?
+	{
+		let mut reader = Cursor::new(
+			KVStore::read(
+				&*kv_store,
+				PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE,
+				PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE,
+				&stored_key,
+			)
+			.await?,
+		);
 		let payment = PaymentDetails::read(&mut reader).map_err(|e| {
 			log_error!(logger, "Failed to deserialize PaymentDetails: {}", e);
 			std::io::Error::new(
@@ -290,17 +310,20 @@ where
 }
 
 /// Read `OutputSweeper` state from the store.
-pub(crate) fn read_output_sweeper(
+pub(crate) async fn read_output_sweeper(
 	broadcaster: Arc<Broadcaster>, fee_estimator: Arc<OnchainFeeEstimator>,
 	chain_data_source: Arc<ChainSource>, keys_manager: Arc<KeysManager>, kv_store: Arc<DynStore>,
 	logger: Arc<Logger>,
 ) -> Result<Sweeper, std::io::Error> {
-	let mut reader = Cursor::new(KVStoreSync::read(
-		&*kv_store,
-		OUTPUT_SWEEPER_PERSISTENCE_PRIMARY_NAMESPACE,
-		OUTPUT_SWEEPER_PERSISTENCE_SECONDARY_NAMESPACE,
-		OUTPUT_SWEEPER_PERSISTENCE_KEY,
-	)?);
+	let mut reader = Cursor::new(
+		KVStore::read(
+			&*kv_store,
+			OUTPUT_SWEEPER_PERSISTENCE_PRIMARY_NAMESPACE,
+			OUTPUT_SWEEPER_PERSISTENCE_SECONDARY_NAMESPACE,
+			OUTPUT_SWEEPER_PERSISTENCE_KEY,
+		)
+		.await?,
+	);
 	let args = (
 		broadcaster,
 		fee_estimator,
@@ -317,18 +340,21 @@ pub(crate) fn read_output_sweeper(
 	Ok(sweeper)
 }
 
-pub(crate) fn read_node_metrics<L: Deref>(
+pub(crate) async fn read_node_metrics<L: Deref>(
 	kv_store: Arc<DynStore>, logger: L,
 ) -> Result<NodeMetrics, std::io::Error>
 where
 	L::Target: LdkLogger,
 {
-	let mut reader = Cursor::new(KVStoreSync::read(
-		&*kv_store,
-		NODE_METRICS_PRIMARY_NAMESPACE,
-		NODE_METRICS_SECONDARY_NAMESPACE,
-		NODE_METRICS_KEY,
-	)?);
+	let mut reader = Cursor::new(
+		KVStore::read(
+			&*kv_store,
+			NODE_METRICS_PRIMARY_NAMESPACE,
+			NODE_METRICS_SECONDARY_NAMESPACE,
+			NODE_METRICS_KEY,
+		)
+		.await?,
+	);
 	NodeMetrics::read(&mut reader).map_err(|e| {
 		log_error!(logger, "Failed to deserialize NodeMetrics: {}", e);
 		std::io::Error::new(std::io::ErrorKind::InvalidData, "Failed to deserialize NodeMetrics")