Implement sync_onchain_wallet for ChainSource::Electrum

Author: tnull

src/chain/electrum.rs

@@ -6,8 +6,8 @@
 // accordance with one or both of these licenses.
 
 use crate::config::{
-	Config, FEE_RATE_CACHE_UPDATE_TIMEOUT_SECS, LDK_WALLET_SYNC_TIMEOUT_SECS,
-	TX_BROADCAST_TIMEOUT_SECS,
+	Config, BDK_CLIENT_STOP_GAP, BDK_WALLET_SYNC_TIMEOUT_SECS, FEE_RATE_CACHE_UPDATE_TIMEOUT_SECS,
+	LDK_WALLET_SYNC_TIMEOUT_SECS, TX_BROADCAST_TIMEOUT_SECS,
 };
 use crate::error::Error;
 use crate::fee_estimator::{
@@ -20,6 +20,12 @@ use lightning::chain::{Confirm, Filter, WatchedOutput};
 use lightning::util::ser::Writeable;
 use lightning_transaction_sync::ElectrumSyncClient;
 
+use bdk_chain::bdk_core::spk_client::FullScanRequest as BdkFullScanRequest;
+use bdk_chain::bdk_core::spk_client::FullScanResponse as BdkFullScanResponse;
+use bdk_chain::bdk_core::spk_client::SyncRequest as BdkSyncRequest;
+use bdk_chain::bdk_core::spk_client::SyncResponse as BdkSyncResponse;
+use bdk_wallet::KeychainKind as BdkKeyChainKind;
+
 use bdk_electrum::BdkElectrumClient;
 
 use electrum_client::{Batch, Client as ElectrumClient, ElectrumApi};
@@ -30,6 +36,8 @@ use std::collections::HashMap;
 use std::sync::Arc;
 use std::time::{Duration, Instant};
 
+const BDK_ELECTRUM_CLIENT_BATCH_SIZE: usize = 5;
+
 pub(crate) struct ElectrumRuntimeClient {
 	electrum_client: Arc<ElectrumClient>,
 	bdk_electrum_client: Arc<BdkElectrumClient<ElectrumClient>>,
@@ -96,6 +104,69 @@ impl ElectrumRuntimeClient {
 		Ok(res)
 	}
 
+	pub(crate) async fn get_full_scan_wallet_update(
+		&self, request: BdkFullScanRequest<BdkKeyChainKind>,
+		cached_txs: impl IntoIterator<Item = impl Into<Arc<Transaction>>>,
+	) -> Result<BdkFullScanResponse<BdkKeyChainKind>, Error> {
+		let bdk_electrum_client = Arc::clone(&self.bdk_electrum_client);
+		bdk_electrum_client.populate_tx_cache(cached_txs);
+
+		let spawn_fut = self.runtime.spawn_blocking(move || {
+			bdk_electrum_client.full_scan(
+				request,
+				BDK_CLIENT_STOP_GAP,
+				BDK_ELECTRUM_CLIENT_BATCH_SIZE,
+				true,
+			)
+		});
+		let wallet_sync_timeout_fut =
+			tokio::time::timeout(Duration::from_secs(BDK_WALLET_SYNC_TIMEOUT_SECS), spawn_fut);
+
+		wallet_sync_timeout_fut
+			.await
+			.map_err(|e| {
+				log_error!(self.logger, "Sync of on-chain wallet timed out: {}", e);
+				Error::WalletOperationTimeout
+			})?
+			.map_err(|e| {
+				log_error!(self.logger, "Sync of on-chain wallet failed: {}", e);
+				Error::WalletOperationFailed
+			})?
+			.map_err(|e| {
+				log_error!(self.logger, "Sync of on-chain wallet failed: {}", e);
+				Error::WalletOperationFailed
+			})
+	}
+
+	pub(crate) async fn get_incremental_sync_wallet_update(
+		&self, request: BdkSyncRequest<(BdkKeyChainKind, u32)>,
+		cached_txs: impl IntoIterator<Item = impl Into<Arc<Transaction>>>,
+	) -> Result<BdkSyncResponse, Error> {
+		let bdk_electrum_client = Arc::clone(&self.bdk_electrum_client);
+		bdk_electrum_client.populate_tx_cache(cached_txs);
+
+		let spawn_fut = self.runtime.spawn_blocking(move || {
+			bdk_electrum_client.sync(request, BDK_ELECTRUM_CLIENT_BATCH_SIZE, true)
+		});
+		let wallet_sync_timeout_fut =
+			tokio::time::timeout(Duration::from_secs(BDK_WALLET_SYNC_TIMEOUT_SECS), spawn_fut);
+
+		wallet_sync_timeout_fut
+			.await
+			.map_err(|e| {
+				log_error!(self.logger, "Incremental sync of on-chain wallet timed out: {}", e);
+				Error::WalletOperationTimeout
+			})?
+			.map_err(|e| {
+				log_error!(self.logger, "Incremental sync of on-chain wallet failed: {}", e);
+				Error::WalletOperationFailed
+			})?
+			.map_err(|e| {
+				log_error!(self.logger, "Incremental sync of on-chain wallet failed: {}", e);
+				Error::WalletOperationFailed
+			})
+	}
+
 	pub(crate) async fn broadcast(&self, tx: Transaction) {
 		let electrum_client = Arc::clone(&self.electrum_client);
 

src/chain/mod.rs

@@ -658,7 +658,95 @@ impl ChainSource {
 
 				res
 			},
-			Self::Electrum { .. } => todo!(),
+			Self::Electrum {
+				electrum_runtime_client,
+				onchain_wallet,
+				onchain_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 onchain wallet"
+						);
+						return Err(Error::FeerateEstimationUpdateFailed);
+					};
+				let receiver_res = {
+					let mut status_lock = onchain_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
+					})?;
+				}
+
+				// If this is our first sync, do a full scan with the configured gap limit.
+				// Otherwise just do an incremental sync.
+				let incremental_sync =
+					node_metrics.read().unwrap().latest_onchain_wallet_sync_timestamp.is_some();
+
+				macro_rules! get_and_apply_wallet_update {
+					($sync_future: expr) => {{
+						let now = Instant::now();
+						let update = $sync_future.await?;
+
+						match onchain_wallet.apply_update(update) {
+							Ok(()) => {
+								log_info!(
+									logger,
+									"{} of on-chain wallet finished in {}ms.",
+									if incremental_sync { "Incremental sync" } else { "Sync" },
+									now.elapsed().as_millis()
+								);
+								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_onchain_wallet_sync_timestamp =
+										unix_time_secs_opt;
+									write_node_metrics(
+										&*locked_node_metrics,
+										Arc::clone(&kv_store),
+										Arc::clone(&logger),
+									)?;
+								}
+								Ok(())
+							},
+							Err(e) => Err(e),
+						}
+					}};
+				}
+
+				let cached_txs = onchain_wallet.get_cached_txs();
+
+				let res = if incremental_sync {
+					let incremental_sync_request = onchain_wallet.get_incremental_sync_request();
+					let incremental_sync_fut = electrum_client
+						.get_incremental_sync_wallet_update(incremental_sync_request, cached_txs);
+					get_and_apply_wallet_update!(incremental_sync_fut)
+				} else {
+					let full_scan_request = onchain_wallet.get_full_scan_request();
+					let full_scan_fut =
+						electrum_client.get_full_scan_wallet_update(full_scan_request, cached_txs);
+					get_and_apply_wallet_update!(full_scan_fut)
+				};
+
+				onchain_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.

src/wallet/mod.rs

@@ -98,6 +98,10 @@ where
 		self.inner.lock().unwrap().start_sync_with_revealed_spks().build()
 	}
 
+	pub(crate) fn get_cached_txs(&self) -> Vec<Arc<Transaction>> {
+		self.inner.lock().unwrap().tx_graph().full_txs().map(|tx_node| tx_node.tx).collect()
+	}
+
 	pub(crate) fn current_best_block(&self) -> BestBlock {
 		let checkpoint = self.inner.lock().unwrap().latest_checkpoint();
 		BestBlock { block_hash: checkpoint.hash(), height: checkpoint.height() }