Give Wallet access to EventQueue

Let wallet sync enqueue on-chain payment events directly.

Co-Authored-By: HAL 9000

Author: tnull

src/builder.rs

@@ -1440,8 +1440,8 @@ fn build_with_store_internal(
 
 	let kv_store_ref = Arc::clone(&kv_store);
 	let logger_ref = Arc::clone(&logger);
-	let (payment_store_res, node_metris_res, pending_payment_store_res) =
-		runtime.block_on(async move {
+	let (payment_store_res, node_metris_res, pending_payment_store_res, event_queue_res) = runtime
+		.block_on(async move {
 			tokio::join!(
 				read_all_objects(
 					&*kv_store_ref,
@@ -1455,7 +1455,8 @@ fn build_with_store_internal(
 					PENDING_PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE,
 					PENDING_PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE,
 					Arc::clone(&logger_ref),
-				)
+				),
+				read_event_queue(Arc::clone(&kv_store_ref), Arc::clone(&logger_ref)),
 			)
 		});
 
@@ -1742,6 +1743,18 @@ fn build_with_store_internal(
 		},
 	};
 
+	let event_queue = match event_queue_res {
+		Ok(event_queue) => Arc::new(event_queue),
+		Err(e) => {
+			if e.kind() == std::io::ErrorKind::NotFound {
+				Arc::new(EventQueue::new(Arc::clone(&kv_store), Arc::clone(&logger)))
+			} else {
+				log_error!(logger, "Failed to read event queue from store: {}", e);
+				return Err(BuildError::ReadFailed);
+			}
+		},
+	};
+
 	let wallet = Arc::new(Wallet::new(
 		bdk_wallet,
 		wallet_persister,
@@ -1751,6 +1764,7 @@ fn build_with_store_internal(
 		Arc::clone(&payment_store),
 		Arc::clone(&runtime),
 		Arc::clone(&config),
+		Arc::clone(&event_queue),
 		Arc::clone(&logger),
 		Arc::clone(&pending_payment_store),
 	));
@@ -1853,7 +1867,6 @@ fn build_with_store_internal(
 		external_scores_res,
 		channel_manager_bytes_res,
 		sweeper_bytes_res,
-		event_queue_res,
 		peer_info_res,
 	) = runtime.block_on(async move {
 		tokio::join!(
@@ -1866,7 +1879,6 @@ fn build_with_store_internal(
 				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)),
 		)
 	});
@@ -2212,18 +2224,6 @@ fn build_with_store_internal(
 		},
 	};
 
-	let event_queue = match event_queue_res {
-		Ok(event_queue) => Arc::new(event_queue),
-		Err(e) => {
-			if e.kind() == std::io::ErrorKind::NotFound {
-				Arc::new(EventQueue::new(Arc::clone(&kv_store), Arc::clone(&logger)))
-			} else {
-				log_error!(logger, "Failed to read event queue from store: {}", e);
-				return Err(BuildError::ReadFailed);
-			}
-		},
-	};
-
 	let peer_store = match peer_info_res {
 		Ok(peer_store) => Arc::new(peer_store),
 		Err(e) => {

src/wallet/mod.rs

@@ -54,6 +54,7 @@ use lightning_invoice::RawBolt11Invoice;
 use persist::KVStoreWalletPersister;
 
 use crate::config::Config;
+use crate::event::EventQueue;
 use crate::fee_estimator::{ConfirmationTarget, FeeEstimator, OnchainFeeEstimator};
 use crate::logger::{log_debug, log_error, log_info, log_trace, LdkLogger, Logger};
 use crate::payment::store::ConfirmationStatus;
@@ -91,6 +92,7 @@ pub(crate) struct Wallet {
 	payment_store: Arc<PaymentStore>,
 	runtime: Arc<Runtime>,
 	config: Arc<Config>,
+	event_queue: Arc<EventQueue<Arc<Logger>>>,
 	logger: Arc<Logger>,
 	pending_payment_store: Arc<PendingPaymentStore>,
 }
@@ -101,7 +103,8 @@ impl Wallet {
 		wallet_persister: KVStoreWalletPersister, broadcaster: Arc<Broadcaster>,
 		fee_estimator: Arc<OnchainFeeEstimator>, chain_source: Arc<ChainSource>,
 		payment_store: Arc<PaymentStore>, runtime: Arc<Runtime>, config: Arc<Config>,
-		logger: Arc<Logger>, pending_payment_store: Arc<PendingPaymentStore>,
+		event_queue: Arc<EventQueue<Arc<Logger>>>, logger: Arc<Logger>,
+		pending_payment_store: Arc<PendingPaymentStore>,
 	) -> Self {
 		let inner = Mutex::new(wallet);
 		let persister = Mutex::new(wallet_persister);
@@ -114,6 +117,7 @@ impl Wallet {
 			payment_store,
 			runtime,
 			config,
+			event_queue,
 			logger,
 			pending_payment_store,
 		}