Implement sync_lightning_wallet for ChainSource::Electrum

Author: tnull

src/chain/electrum.rs

@@ -5,15 +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::{Config, FEE_RATE_CACHE_UPDATE_TIMEOUT_SECS, TX_BROADCAST_TIMEOUT_SECS};
+use crate::config::{
+	Config, FEE_RATE_CACHE_UPDATE_TIMEOUT_SECS, LDK_WALLET_SYNC_TIMEOUT_SECS,
+	TX_BROADCAST_TIMEOUT_SECS,
+};
 use crate::error::Error;
 use crate::fee_estimator::{
 	apply_post_estimation_adjustments, get_all_conf_targets, get_num_block_defaults_for_target,
 	ConfirmationTarget,
 };
-use crate::logger::{log_bytes, log_error, log_trace, LdkLogger, Logger};
+use crate::logger::{log_bytes, log_error, log_info, log_trace, LdkLogger, Logger};
 
-use lightning::chain::{Filter, WatchedOutput};
+use lightning::chain::{Confirm, Filter, WatchedOutput};
 use lightning::util::ser::Writeable;
 use lightning_transaction_sync::ElectrumSyncClient;
 
@@ -25,7 +28,7 @@ use bitcoin::{FeeRate, Network, Script, Transaction, Txid};
 
 use std::collections::HashMap;
 use std::sync::Arc;
-use std::time::Duration;
+use std::time::{Duration, Instant};
 
 pub(crate) struct ElectrumRuntimeClient {
 	electrum_client: Arc<ElectrumClient>,
@@ -59,6 +62,40 @@ impl ElectrumRuntimeClient {
 		Ok(Self { electrum_client, bdk_electrum_client, tx_sync, runtime, config, logger })
 	}
 
+	pub(crate) async fn sync_confirmables(
+		&self, confirmables: Vec<Arc<dyn Confirm + Sync + Send>>,
+	) -> Result<(), Error> {
+		let now = Instant::now();
+
+		let tx_sync = Arc::clone(&self.tx_sync);
+		let spawn_fut = self.runtime.spawn_blocking(move || tx_sync.sync(confirmables));
+		let timeout_fut =
+			tokio::time::timeout(Duration::from_secs(LDK_WALLET_SYNC_TIMEOUT_SECS), spawn_fut);
+
+		let res = timeout_fut
+			.await
+			.map_err(|e| {
+				log_error!(self.logger, "Sync of Lightning wallet timed out: {}", e);
+				Error::TxSyncTimeout
+			})?
+			.map_err(|e| {
+				log_error!(self.logger, "Sync of Lightning wallet failed: {}", e);
+				Error::TxSyncFailed
+			})?
+			.map_err(|e| {
+				log_error!(self.logger, "Sync of Lightning wallet failed: {}", e);
+				Error::TxSyncFailed
+			})?;
+
+		log_info!(
+			self.logger,
+			"Sync of Lightning wallet finished in {}ms.",
+			now.elapsed().as_millis()
+		);
+
+		Ok(res)
+	}
+
 	pub(crate) async fn broadcast(&self, tx: Transaction) {
 		let electrum_client = Arc::clone(&self.electrum_client);
 

src/chain/mod.rs

@@ -758,7 +758,76 @@ impl ChainSource {
 
 				res
 			},
-			Self::Electrum { .. } => todo!(),
+			Self::Electrum {
+				electrum_runtime_client,
+				lightning_wallet_sync_status,
+				kv_store,
+				logger,
+				node_metrics,
+				..
+			} => {
+				let electrum_client: Arc<ElectrumRuntimeClient> =
+					if let Some(client) = electrum_runtime_client.read().unwrap().as_ref() {
+						Arc::clone(client)
+					} else {
+						debug_assert!(
+							false,
+							"We should have started the chain source before syncing the lightning wallet"
+						);
+						return Err(Error::FeerateEstimationUpdateFailed);
+					};
+
+				let sync_cman = Arc::clone(&channel_manager);
+				let sync_cmon = Arc::clone(&chain_monitor);
+				let sync_sweeper = Arc::clone(&output_sweeper);
+				let confirmables = vec![
+					sync_cman as Arc<dyn Confirm + Sync + Send>,
+					sync_cmon as Arc<dyn Confirm + Sync + Send>,
+					sync_sweeper as Arc<dyn Confirm + Sync + Send>,
+				];
+
+				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 = electrum_client.sync_confirmables(confirmables).await;
+
+				if let Ok(_) = res {
+					let unix_time_secs_opt =
+						SystemTime::now().duration_since(UNIX_EPOCH).ok().map(|d| d.as_secs());
+					{
+						let mut locked_node_metrics = node_metrics.write().unwrap();
+						locked_node_metrics.latest_lightning_wallet_sync_timestamp =
+							unix_time_secs_opt;
+						write_node_metrics(
+							&*locked_node_metrics,
+							Arc::clone(&kv_store),
+							Arc::clone(&logger),
+						)?;
+					}
+
+					periodically_archive_fully_resolved_monitors(
+						Arc::clone(&channel_manager),
+						Arc::clone(&chain_monitor),
+						Arc::clone(&kv_store),
+						Arc::clone(&logger),
+						Arc::clone(&node_metrics),
+					)?;
+				}
+
+				lightning_wallet_sync_status.lock().unwrap().propagate_result_to_subscribers(res);
+
+				res
+			},
 			Self::BitcoindRpc { .. } => {
 				// In BitcoindRpc mode we sync lightning and onchain wallet in one go by via
 				// `ChainPoller`. So nothing to do here.