Merge upstream/main

Author: ben-kaufman

src/chain/electrum.rs

@@ -14,6 +14,7 @@ use bdk_chain::bdk_core::spk_client::{
 	SyncRequest as BdkSyncRequest, SyncResponse as BdkSyncResponse,
 };
 use bdk_electrum::BdkElectrumClient;
+use bdk_wallet::event::WalletEvent;
 use bdk_wallet::{KeychainKind as BdkKeyChainKind, Update as BdkUpdate};
 use bitcoin::{FeeRate, Network, Script, ScriptBuf, Transaction, Txid};
 use electrum_client::{
@@ -39,8 +40,6 @@ use crate::runtime::Runtime;
 use crate::types::{ChainMonitor, ChannelManager, DynStore, Sweeper, Wallet};
 use crate::NodeMetrics;
 
-use bdk_wallet::event::WalletEvent;
-
 const BDK_ELECTRUM_CLIENT_BATCH_SIZE: usize = 5;
 const ELECTRUM_CLIENT_NUM_RETRIES: u8 = 3;
 const ELECTRUM_CLIENT_TIMEOUT_SECS: u8 = 10;
@@ -103,21 +102,30 @@ impl ElectrumChainSource {
 		};
 		if let Some(mut sync_receiver) = receiver_res {
 			log_info!(self.logger, "Sync in progress, skipping.");
-			return sync_receiver.recv().await.map(|res| res.map(|_| Vec::new())).map_err(|e| {
-				debug_assert!(false, "Failed to receive wallet sync result: {:?}", e);
-				log_error!(self.logger, "Failed to receive wallet sync result: {:?}", e);
-				Error::WalletOperationFailed
-			})?;
+			match sync_receiver.recv().await {
+				Ok(Ok(())) => return Ok(Vec::new()),
+				Ok(Err(e)) => return Err(e),
+				Err(e) => {
+					debug_assert!(false, "Failed to receive wallet sync result: {:?}", e);
+					log_error!(self.logger, "Failed to receive wallet sync result: {:?}", e);
+					return Err(Error::WalletOperationFailed);
+				},
+			}
 		}
 
 		let res = self.sync_onchain_wallet_inner(onchain_wallet).await;
 
-		self.onchain_wallet_sync_status.lock().unwrap().propagate_result_to_subscribers(res.as_ref().map(|_| ()).map_err(|e| e.clone()));
+		self.onchain_wallet_sync_status
+			.lock()
+			.unwrap()
+			.propagate_result_to_subscribers(res.as_ref().map(|_| ()).map_err(|e| e.clone()));
 
 		res
 	}
 
-	async fn sync_onchain_wallet_inner(&self, onchain_wallet: Arc<Wallet>) -> Result<Vec<WalletEvent>, Error> {
+	async fn sync_onchain_wallet_inner(
+		&self, onchain_wallet: Arc<Wallet>,
+	) -> Result<Vec<WalletEvent>, Error> {
 		let electrum_client: Arc<ElectrumRuntimeClient> =
 			if let Some(client) = self.electrum_runtime_status.read().unwrap().client().as_ref() {
 				Arc::clone(client)

src/chain/esplora.rs

@@ -10,6 +10,7 @@ use std::sync::{Arc, Mutex, RwLock};
 use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
 
 use bdk_esplora::EsploraAsyncExt;
+use bdk_wallet::event::WalletEvent;
 use bitcoin::{FeeRate, Network, Script, Transaction, Txid};
 use esplora_client::AsyncClient as EsploraAsyncClient;
 use lightning::chain::{Confirm, Filter, WatchedOutput};
@@ -31,8 +32,6 @@ use crate::logger::{log_bytes, log_error, log_info, log_trace, LdkLogger, Logger
 use crate::types::{ChainMonitor, ChannelManager, DynStore, Sweeper, Wallet};
 use crate::{Error, NodeMetrics};
 
-use bdk_wallet::event::WalletEvent;
-
 pub(super) struct EsploraChainSource {
 	pub(super) sync_config: EsploraSyncConfig,
 	esplora_client: EsploraAsyncClient,
@@ -88,23 +87,30 @@ impl EsploraChainSource {
 		};
 		if let Some(mut sync_receiver) = receiver_res {
 			log_info!(self.logger, "Sync in progress, skipping.");
-			sync_receiver.recv().await.map_err(|e| {
-				debug_assert!(false, "Failed to receive wallet sync result: {:?}", e);
-				log_error!(self.logger, "Failed to receive wallet sync result: {:?}", e);
-				Error::WalletOperationFailed
-			})??;
-			// Return empty events since we're just waiting for another sync
-			return Ok(Vec::new());
+			match sync_receiver.recv().await {
+				Ok(Ok(())) => return Ok(Vec::new()),
+				Ok(Err(e)) => return Err(e),
+				Err(e) => {
+					debug_assert!(false, "Failed to receive wallet sync result: {:?}", e);
+					log_error!(self.logger, "Failed to receive wallet sync result: {:?}", e);
+					return Err(Error::WalletOperationFailed);
+				},
+			}
 		}
 
 		let res = self.sync_onchain_wallet_inner(onchain_wallet).await;
 
-		self.onchain_wallet_sync_status.lock().unwrap().propagate_result_to_subscribers(res.as_ref().map(|_| ()).map_err(|e| e.clone()));
+		self.onchain_wallet_sync_status
+			.lock()
+			.unwrap()
+			.propagate_result_to_subscribers(res.as_ref().map(|_| ()).map_err(|e| e.clone()));
 
 		res
 	}
 
-	async fn sync_onchain_wallet_inner(&self, onchain_wallet: Arc<Wallet>) -> Result<Vec<WalletEvent>, Error> {
+	async fn sync_onchain_wallet_inner(
+		&self, onchain_wallet: Arc<Wallet>,
+	) -> Result<Vec<WalletEvent>, Error> {
 		// If this is our first sync, do a full scan with the configured gap limit.
 		// Otherwise just do an incremental sync.
 		let incremental_sync =

src/chain/mod.rs

@@ -256,7 +256,8 @@ where
 				// We don't emit an event for chain tip changes as this is too noisy
 			},
 			BdkWalletEvent::TxReplaced { txid, conflicts, .. } => {
-				let conflict_txids: Vec<Txid> = conflicts.iter().map(|(_, conflict_txid)| *conflict_txid).collect();
+				let conflict_txids: Vec<Txid> =
+					conflicts.iter().map(|(_, conflict_txid)| *conflict_txid).collect();
 				log_info!(
 					logger,
 					"Onchain transaction {} was replaced by {} transaction(s)",
@@ -295,7 +296,16 @@ impl ChainSource {
 			node_metrics,
 		);
 		let kind = ChainSourceKind::Esplora(esplora_chain_source);
-		(Self { kind, tx_broadcaster, logger, onchain_wallet: Arc::new(Mutex::new(None)), event_queue: Arc::new(Mutex::new(None)) }, None)
+		(
+			Self {
+				kind,
+				tx_broadcaster,
+				logger,
+				onchain_wallet: Arc::new(Mutex::new(None)),
+				event_queue: Arc::new(Mutex::new(None)),
+			},
+			None,
+		)
 	}
 
 	pub(crate) fn new_electrum(
@@ -314,7 +324,16 @@ impl ChainSource {
 			node_metrics,
 		);
 		let kind = ChainSourceKind::Electrum(electrum_chain_source);
-		(Self { kind, tx_broadcaster, logger, onchain_wallet: Arc::new(Mutex::new(None)), event_queue: Arc::new(Mutex::new(None)) }, None)
+		(
+			Self {
+				kind,
+				tx_broadcaster,
+				logger,
+				onchain_wallet: Arc::new(Mutex::new(None)),
+				event_queue: Arc::new(Mutex::new(None)),
+			},
+			None,
+		)
 	}
 
 	pub(crate) async fn new_bitcoind_rpc(
@@ -336,7 +355,16 @@ impl ChainSource {
 		);
 		let best_block = bitcoind_chain_source.poll_best_block().await.ok();
 		let kind = ChainSourceKind::Bitcoind(bitcoind_chain_source);
-		(Self { kind, tx_broadcaster, logger, onchain_wallet: Arc::new(Mutex::new(None)), event_queue: Arc::new(Mutex::new(None)) }, best_block)
+		(
+			Self {
+				kind,
+				tx_broadcaster,
+				logger,
+				onchain_wallet: Arc::new(Mutex::new(None)),
+				event_queue: Arc::new(Mutex::new(None)),
+			},
+			best_block,
+		)
 	}
 
 	pub(crate) async fn new_bitcoind_rest(
@@ -359,7 +387,16 @@ impl ChainSource {
 		);
 		let best_block = bitcoind_chain_source.poll_best_block().await.ok();
 		let kind = ChainSourceKind::Bitcoind(bitcoind_chain_source);
-		(Self { kind, tx_broadcaster, logger, onchain_wallet: Arc::new(Mutex::new(None)), event_queue: Arc::new(Mutex::new(None)) }, best_block)
+		(
+			Self {
+				kind,
+				tx_broadcaster,
+				logger,
+				onchain_wallet: Arc::new(Mutex::new(None)),
+				event_queue: Arc::new(Mutex::new(None)),
+			},
+			best_block,
+		)
 	}
 
 	pub(crate) fn start(&self, runtime: Arc<Runtime>) -> Result<(), Error> {
@@ -468,14 +505,17 @@ impl ChainSource {
 
 	// Synchronize the onchain wallet via transaction-based protocols (Esplora, Electrum).
 	// If event_queue is set, emits onchain events.
-	pub(crate) async fn sync_onchain_wallet(&self, wallet: Arc<Wallet>) -> Result<(), Error> {
+	pub(crate) async fn sync_onchain_wallet(
+		&self, wallet: Arc<Wallet>, channel_manager: Option<Arc<ChannelManager>>,
+		chain_monitor: Option<Arc<ChainMonitor>>,
+	) -> Result<(), Error> {
 		let event_queue = self.event_queue.lock().unwrap().clone();
 		if let Some(event_queue) = event_queue {
 			// Use event-emitting sync path
 			self.sync_onchain_wallet_with_events(
 				Some(&event_queue),
-				None,
-				None,
+				channel_manager.as_ref(),
+				chain_monitor.as_ref(),
 				self.config(),
 			)
 			.await
@@ -584,8 +624,8 @@ impl ChainSource {
 	// etc.) with event emission support.
 	async fn sync_onchain_wallet_with_events(
 		&self, event_queue: Option<&Arc<EventQueue<Arc<Logger>>>>,
-		channel_manager: Option<&Arc<ChannelManager>>,
-		chain_monitor: Option<&Arc<ChainMonitor>>, config: Option<&Arc<Config>>,
+		channel_manager: Option<&Arc<ChannelManager>>, chain_monitor: Option<&Arc<ChainMonitor>>,
+		config: Option<&Arc<Config>>,
 	) -> Result<(), Error> {
 		let wallet = self.onchain_wallet.lock().unwrap().clone();
 		let wallet = wallet.ok_or(Error::WalletOperationFailed)?;
@@ -594,9 +634,10 @@ impl ChainSource {
 			ChainSourceKind::Esplora(esplora_chain_source) => {
 				// Track unconfirmed transactions before sync to detect evictions
 				let prev_unconfirmed_txids = wallet.get_unconfirmed_txids();
-				
-				let wallet_events = esplora_chain_source.sync_onchain_wallet(Arc::clone(&wallet)).await?;
-				
+
+				let wallet_events =
+					esplora_chain_source.sync_onchain_wallet(Arc::clone(&wallet)).await?;
+
 				// Process wallet events if event queue is provided
 				if let Some(event_queue) = event_queue {
 					process_wallet_events(
@@ -606,33 +647,41 @@ impl ChainSource {
 						&self.logger,
 						channel_manager,
 						chain_monitor,
-					).await?;
+					)
+					.await?;
 
 					// Check for evicted transactions
 					check_and_emit_evicted_transactions(
 						prev_unconfirmed_txids,
 						&wallet,
 						event_queue,
 						&self.logger,
-					).await;
+					)
+					.await;
 
 					// Emit SyncCompleted event
 					let synced_height = wallet.current_best_block().height;
-					event_queue.add_event(Event::SyncCompleted {
-						sync_type: SyncType::OnchainWallet,
-						synced_block_height: synced_height,
-					}).await?;
+					event_queue
+						.add_event(Event::SyncCompleted {
+							sync_type: SyncType::OnchainWallet,
+							synced_block_height: synced_height,
+						})
+						.await?;
 					// Check for balance changes and emit BalanceChanged event if needed
-					if let (Some(cm), Some(chain_mon), Some(cfg)) = (channel_manager, chain_monitor, config) {
-						let cur_anchor_reserve_sats = crate::total_anchor_channels_reserve_sats(cm, cfg);
+					if let (Some(cm), Some(chain_mon), Some(cfg)) =
+						(channel_manager, chain_monitor, config)
+					{
+						let cur_anchor_reserve_sats =
+							crate::total_anchor_channels_reserve_sats(cm, cfg);
 						let (total_onchain_balance_sats, spendable_onchain_balance_sats) =
 							wallet.get_balances(cur_anchor_reserve_sats).unwrap_or((0, 0));
 
 						let mut total_lightning_balance_sats = 0;
 						for channel_id in chain_mon.list_monitors() {
 							if let Ok(monitor) = chain_mon.get_monitor(channel_id) {
 								for ldk_balance in monitor.get_claimable_balances() {
-									total_lightning_balance_sats += ldk_balance.claimable_amount_satoshis();
+									total_lightning_balance_sats +=
+										ldk_balance.claimable_amount_satoshis();
 								}
 							}
 						}
@@ -642,7 +691,7 @@ impl ChainSource {
 							spendable_onchain_balance_sats,
 							total_anchor_channels_reserve_sats: std::cmp::min(
 								cur_anchor_reserve_sats,
-								total_onchain_balance_sats
+								total_onchain_balance_sats,
 							),
 							total_lightning_balance_sats,
 							lightning_balances: Vec::new(),
@@ -666,17 +715,19 @@ impl ChainSource {
 							event_queue,
 							&kv_store,
 							&self.logger,
-						).await?;
+						)
+						.await?;
 					}
 				}
 				Ok(())
 			},
 			ChainSourceKind::Electrum(electrum_chain_source) => {
 				// Track unconfirmed transactions before sync to detect evictions
 				let prev_unconfirmed_txids = wallet.get_unconfirmed_txids();
-				
-				let wallet_events = electrum_chain_source.sync_onchain_wallet(Arc::clone(&wallet)).await?;
-				
+
+				let wallet_events =
+					electrum_chain_source.sync_onchain_wallet(Arc::clone(&wallet)).await?;
+
 				// Process wallet events if event queue is provided
 				if let Some(event_queue) = event_queue {
 					process_wallet_events(
@@ -686,34 +737,42 @@ impl ChainSource {
 						&self.logger,
 						channel_manager,
 						chain_monitor,
-					).await?;
+					)
+					.await?;
 
 					// Check for evicted transactions
 					check_and_emit_evicted_transactions(
 						prev_unconfirmed_txids,
 						&wallet,
 						event_queue,
 						&self.logger,
-					).await;
+					)
+					.await;
 
 					// Emit SyncCompleted event
 					let synced_height = wallet.current_best_block().height;
-					event_queue.add_event(Event::SyncCompleted {
-						sync_type: SyncType::OnchainWallet,
-						synced_block_height: synced_height,
-					}).await?;
+					event_queue
+						.add_event(Event::SyncCompleted {
+							sync_type: SyncType::OnchainWallet,
+							synced_block_height: synced_height,
+						})
+						.await?;
 
 					// Check for balance changes and emit BalanceChanged event if needed
-					if let (Some(cm), Some(chain_mon), Some(cfg)) = (channel_manager, chain_monitor, config) {
-						let cur_anchor_reserve_sats = crate::total_anchor_channels_reserve_sats(cm, cfg);
+					if let (Some(cm), Some(chain_mon), Some(cfg)) =
+						(channel_manager, chain_monitor, config)
+					{
+						let cur_anchor_reserve_sats =
+							crate::total_anchor_channels_reserve_sats(cm, cfg);
 						let (total_onchain_balance_sats, spendable_onchain_balance_sats) =
 							wallet.get_balances(cur_anchor_reserve_sats).unwrap_or((0, 0));
 
 						let mut total_lightning_balance_sats = 0;
 						for channel_id in chain_mon.list_monitors() {
 							if let Ok(monitor) = chain_mon.get_monitor(channel_id) {
 								for ldk_balance in monitor.get_claimable_balances() {
-									total_lightning_balance_sats += ldk_balance.claimable_amount_satoshis();
+									total_lightning_balance_sats +=
+										ldk_balance.claimable_amount_satoshis();
 								}
 							}
 						}
@@ -723,7 +782,7 @@ impl ChainSource {
 							spendable_onchain_balance_sats,
 							total_anchor_channels_reserve_sats: std::cmp::min(
 								cur_anchor_reserve_sats,
-								total_onchain_balance_sats
+								total_onchain_balance_sats,
 							),
 							total_lightning_balance_sats,
 							lightning_balances: Vec::new(),
@@ -747,7 +806,8 @@ impl ChainSource {
 							event_queue,
 							&kv_store,
 							&self.logger,
-						).await?;
+						)
+						.await?;
 					}
 				}
 				Ok(())