Move Lightning wallet syncing to ChainSource

Author: tnull

src/builder.rs

@@ -50,8 +50,6 @@ use lightning::util::sweep::OutputSweeper;
 
 use lightning_persister::fs_store::FilesystemStore;
 
-use lightning_transaction_sync::EsploraSyncClient;
-
 use lightning_liquidity::lsps2::client::LSPS2ClientConfig;
 use lightning_liquidity::{LiquidityClientConfig, LiquidityManager};
 
@@ -562,82 +560,71 @@ fn build_with_store_internal(
 			})?,
 	};
 
-	let (wallet, chain_source, tx_sync, tx_broadcaster, fee_estimator) =
-		match chain_data_source_config {
-			Some(ChainDataSourceConfig::Esplora(server_url)) => {
-				let mut client_builder = esplora_client::Builder::new(&server_url.clone());
-				client_builder = client_builder.timeout(DEFAULT_ESPLORA_CLIENT_TIMEOUT_SECS);
-				let esplora_client = client_builder.build_async().unwrap();
-				let tx_sync =
-					Arc::new(EsploraSyncClient::from_client(esplora_client, Arc::clone(&logger)));
-				let tx_broadcaster = Arc::new(TransactionBroadcaster::new(
-					tx_sync.client().clone(),
-					Arc::clone(&logger),
-				));
-				let fee_estimator = Arc::new(OnchainFeeEstimator::new(
-					tx_sync.client().clone(),
-					Arc::clone(&config),
-					Arc::clone(&logger),
-				));
+	let (wallet, chain_source, tx_broadcaster, fee_estimator) = match chain_data_source_config {
+		Some(ChainDataSourceConfig::Esplora(server_url)) => {
+			let mut client_builder = esplora_client::Builder::new(&server_url.clone());
+			client_builder = client_builder.timeout(DEFAULT_ESPLORA_CLIENT_TIMEOUT_SECS);
+			let esplora_client = client_builder.build_async().unwrap();
+			let tx_broadcaster =
+				Arc::new(TransactionBroadcaster::new(esplora_client.clone(), Arc::clone(&logger)));
+			let fee_estimator = Arc::new(OnchainFeeEstimator::new(
+				esplora_client,
+				Arc::clone(&config),
+				Arc::clone(&logger),
+			));
 
-				let wallet = Arc::new(Wallet::new(
-					bdk_wallet,
-					wallet_persister,
-					Arc::clone(&tx_broadcaster),
-					Arc::clone(&fee_estimator),
-					Arc::clone(&logger),
-				));
+			let wallet = Arc::new(Wallet::new(
+				bdk_wallet,
+				wallet_persister,
+				Arc::clone(&tx_broadcaster),
+				Arc::clone(&fee_estimator),
+				Arc::clone(&logger),
+			));
 
-				let chain_source = Arc::new(ChainSource::new_esplora(
-					server_url.clone(),
-					Arc::clone(&wallet),
-					Arc::clone(&logger),
-				));
-				(wallet, chain_source, tx_sync, tx_broadcaster, fee_estimator)
-			},
-			None => {
-				// Default to Esplora client.
-				let server_url = DEFAULT_ESPLORA_SERVER_URL.to_string();
-				let mut client_builder = esplora_client::Builder::new(&server_url);
-				client_builder = client_builder.timeout(DEFAULT_ESPLORA_CLIENT_TIMEOUT_SECS);
-				let esplora_client = client_builder.build_async().unwrap();
-				let tx_sync = Arc::new(EsploraSyncClient::from_client(
-					esplora_client.clone(),
-					Arc::clone(&logger),
-				));
-				let tx_broadcaster = Arc::new(TransactionBroadcaster::new(
-					tx_sync.client().clone(),
-					Arc::clone(&logger),
-				));
-				let fee_estimator = Arc::new(OnchainFeeEstimator::new(
-					tx_sync.client().clone(),
-					Arc::clone(&config),
-					Arc::clone(&logger),
-				));
+			let chain_source = Arc::new(ChainSource::new_esplora(
+				server_url.clone(),
+				Arc::clone(&wallet),
+				Arc::clone(&logger),
+			));
+			(wallet, chain_source, tx_broadcaster, fee_estimator)
+		},
+		None => {
+			// Default to Esplora client.
+			let server_url = DEFAULT_ESPLORA_SERVER_URL.to_string();
+			let mut client_builder = esplora_client::Builder::new(&server_url);
+			client_builder = client_builder.timeout(DEFAULT_ESPLORA_CLIENT_TIMEOUT_SECS);
+			let esplora_client = client_builder.build_async().unwrap();
+			let tx_broadcaster =
+				Arc::new(TransactionBroadcaster::new(esplora_client.clone(), Arc::clone(&logger)));
+			let fee_estimator = Arc::new(OnchainFeeEstimator::new(
+				esplora_client,
+				Arc::clone(&config),
+				Arc::clone(&logger),
+			));
 
-				let wallet = Arc::new(Wallet::new(
-					bdk_wallet,
-					wallet_persister,
-					Arc::clone(&tx_broadcaster),
-					Arc::clone(&fee_estimator),
-					Arc::clone(&logger),
-				));
+			let wallet = Arc::new(Wallet::new(
+				bdk_wallet,
+				wallet_persister,
+				Arc::clone(&tx_broadcaster),
+				Arc::clone(&fee_estimator),
+				Arc::clone(&logger),
+			));
 
-				let chain_source = Arc::new(ChainSource::new_esplora(
-					server_url.clone(),
-					Arc::clone(&wallet),
-					Arc::clone(&logger),
-				));
+			let chain_source = Arc::new(ChainSource::new_esplora(
+				server_url.clone(),
+				Arc::clone(&wallet),
+				Arc::clone(&logger),
+			));
 
-				(wallet, chain_source, tx_sync, tx_broadcaster, fee_estimator)
-			},
-		};
+			(wallet, chain_source, tx_broadcaster, fee_estimator)
+		},
+	};
 
 	let runtime = Arc::new(RwLock::new(None));
 
 	// Initialize the ChainMonitor
 	let chain_monitor: Arc<ChainMonitor> = Arc::new(chainmonitor::ChainMonitor::new(
-		Some(Arc::clone(&tx_sync)),
+		Some(Arc::clone(&chain_source)),
 		Arc::clone(&tx_broadcaster),
 		Arc::clone(&logger),
 		Arc::clone(&fee_estimator),
@@ -853,7 +840,7 @@ fn build_with_store_internal(
 			let liquidity_manager = Arc::new(LiquidityManager::new(
 				Arc::clone(&keys_manager),
 				Arc::clone(&channel_manager),
-				Some(Arc::clone(&tx_sync)),
+				Some(Arc::clone(&chain_source)),
 				None,
 				None,
 				liquidity_client_config,
@@ -921,7 +908,7 @@ fn build_with_store_internal(
 	let output_sweeper = match io::utils::read_output_sweeper(
 		Arc::clone(&tx_broadcaster),
 		Arc::clone(&fee_estimator),
-		Arc::clone(&tx_sync),
+		Arc::clone(&chain_source),
 		Arc::clone(&keys_manager),
 		Arc::clone(&kv_store),
 		Arc::clone(&logger),
@@ -933,7 +920,7 @@ fn build_with_store_internal(
 					channel_manager.current_best_block(),
 					Arc::clone(&tx_broadcaster),
 					Arc::clone(&fee_estimator),
-					Some(Arc::clone(&tx_sync)),
+					Some(Arc::clone(&chain_source)),
 					Arc::clone(&keys_manager),
 					Arc::clone(&keys_manager),
 					Arc::clone(&kv_store),
@@ -1010,7 +997,6 @@ fn build_with_store_internal(
 		config,
 		wallet,
 		chain_source,
-		tx_sync,
 		tx_broadcaster,
 		fee_estimator,
 		event_queue,

src/chain/mod.rs

@@ -5,11 +5,18 @@
 // http://opensource.org/licenses/MIT>, at your option. You may not use this file except in
 // accordance with one or both of these licenses.
 
-use crate::config::{BDK_CLIENT_CONCURRENCY, BDK_CLIENT_STOP_GAP, BDK_WALLET_SYNC_TIMEOUT_SECS};
+use crate::config::{
+	BDK_CLIENT_CONCURRENCY, BDK_CLIENT_STOP_GAP, BDK_WALLET_SYNC_TIMEOUT_SECS,
+	LDK_WALLET_SYNC_TIMEOUT_SECS,
+};
 use crate::logger::{log_error, log_info, FilesystemLogger, Logger};
 use crate::types::Wallet;
 use crate::Error;
 
+use lightning::chain::{Confirm, Filter};
+
+use lightning_transaction_sync::EsploraSyncClient;
+
 use bdk_esplora::EsploraAsyncExt;
 
 use esplora_client::AsyncClient as EsploraAsyncClient;
@@ -81,6 +88,8 @@ pub(crate) enum ChainSource {
 		esplora_client: EsploraAsyncClient,
 		onchain_wallet: Arc<Wallet>,
 		onchain_wallet_sync_status: Mutex<WalletSyncStatus>,
+		tx_sync: Arc<EsploraSyncClient<Arc<FilesystemLogger>>>,
+		lightning_wallet_sync_status: Mutex<WalletSyncStatus>,
 		logger: Arc<FilesystemLogger>,
 	},
 }
@@ -92,11 +101,16 @@ impl ChainSource {
 		let mut client_builder = esplora_client::Builder::new(&server_url.clone());
 		client_builder = client_builder.timeout(DEFAULT_ESPLORA_CLIENT_TIMEOUT_SECS);
 		let esplora_client = client_builder.build_async().unwrap();
+		let tx_sync =
+			Arc::new(EsploraSyncClient::from_client(esplora_client.clone(), Arc::clone(&logger)));
 		let onchain_wallet_sync_status = Mutex::new(WalletSyncStatus::Completed);
+		let lightning_wallet_sync_status = Mutex::new(WalletSyncStatus::Completed);
 		Self::Esplora {
 			esplora_client,
 			onchain_wallet,
 			onchain_wallet_sync_status,
+			tx_sync,
+			lightning_wallet_sync_status,
 			logger,
 		}
 	}
@@ -166,4 +180,57 @@ impl ChainSource {
 			},
 		}
 	}
+
+	pub(crate) async fn sync_lightning_wallet(
+		&self, confirmables: Vec<&(dyn Confirm + Send + Sync)>,
+	) -> Result<(), Error> {
+		match self {
+			Self::Esplora { tx_sync, lightning_wallet_sync_status, logger, .. } => {
+				let receiver_res = {
+					let mut status_lock = lightning_wallet_sync_status.lock().unwrap();
+					status_lock.register_or_subscribe_pending_sync()
+				};
+				if let Some(mut sync_receiver) = receiver_res {
+					log_info!(logger, "Sync in progress, skipping.");
+					return sync_receiver.recv().await.map_err(|e| {
+						debug_assert!(false, "Failed to receive wallet sync result: {:?}", e);
+						log_error!(logger, "Failed to receive wallet sync result: {:?}", e);
+						Error::WalletOperationFailed
+					})?;
+				}
+				let res = {
+					let timeout_fut = tokio::time::timeout(
+						Duration::from_secs(LDK_WALLET_SYNC_TIMEOUT_SECS),
+						tx_sync.sync(confirmables),
+					);
+					match timeout_fut.await {
+						Ok(res) => res.map_err(|_| Error::TxSyncFailed),
+						Err(e) => {
+							log_error!(logger, "Lightning wallet sync timed out: {}", e);
+							Err(Error::TxSyncTimeout)
+						},
+					}
+				};
+
+				lightning_wallet_sync_status.lock().unwrap().propagate_result_to_subscribers(res);
+
+				res
+			},
+		}
+	}
+}
+
+impl Filter for ChainSource {
+	fn register_tx(&self, txid: &bitcoin::Txid, script_pubkey: &bitcoin::Script) {
+		{
+			match self {
+				Self::Esplora { tx_sync, .. } => tx_sync.register_tx(txid, script_pubkey),
+			}
+		}
+	}
+	fn register_output(&self, output: lightning::chain::WatchedOutput) {
+		match self {
+			Self::Esplora { tx_sync, .. } => tx_sync.register_output(output),
+		}
+	}
 }

src/io/utils.rs

@@ -8,10 +8,11 @@
 use super::*;
 use crate::config::WALLET_KEYS_SEED_LEN;
 
+use crate::chain::ChainSource;
 use crate::logger::{log_error, FilesystemLogger};
 use crate::peer_store::PeerStore;
 use crate::sweep::DeprecatedSpendableOutputInfo;
-use crate::types::{Broadcaster, ChainSource, DynStore, FeeEstimator, KeysManager, Sweeper};
+use crate::types::{Broadcaster, DynStore, FeeEstimator, KeysManager, Sweeper};
 use crate::wallet::ser::{ChangeSetDeserWrapper, ChangeSetSerWrapper};
 use crate::{Error, EventQueue, PaymentDetails};