Implement (cached) Filter for ElectrumRuntimeClient

Currently, we won't have a `Runtime` available when initializing
`ChainSource::Electrum`. We therefore isolate any runtime-specific
behavior into an `ElectrumRuntimeClient`.

Here, we implement `Filter` for `ElectrumRuntimeClient`, but we need to
cache the registrations as they might happen prior to
`ElectrumRuntimeClient` becoming available.

Author: tnull

src/chain/electrum.rs

@@ -8,12 +8,15 @@
 use crate::error::Error;
 use crate::logger::{log_error, LdkLogger, Logger};
 
+use lightning::chain::{Filter, WatchedOutput};
 use lightning_transaction_sync::ElectrumSyncClient;
 
 use bdk_electrum::BdkElectrumClient;
 
 use electrum_client::Client as ElectrumClient;
 
+use bitcoin::{Script, Txid};
+
 use std::sync::Arc;
 
 pub(crate) struct ElectrumRuntimeClient {
@@ -46,3 +49,12 @@ impl ElectrumRuntimeClient {
 		Ok(Self { electrum_client, bdk_electrum_client, tx_sync, runtime, logger })
 	}
 }
+
+impl Filter for ElectrumRuntimeClient {
+	fn register_tx(&self, txid: &Txid, script_pubkey: &Script) {
+		self.tx_sync.register_tx(txid, script_pubkey)
+	}
+	fn register_output(&self, output: WatchedOutput) {
+		self.tx_sync.register_output(output)
+	}
+}

src/chain/mod.rs

@@ -28,7 +28,7 @@ use crate::types::{Broadcaster, ChainMonitor, ChannelManager, DynStore, Sweeper,
 use crate::{Error, NodeMetrics};
 
 use lightning::chain::chaininterface::ConfirmationTarget as LdkConfirmationTarget;
-use lightning::chain::{Confirm, Filter, Listen};
+use lightning::chain::{Confirm, Filter, Listen, WatchedOutput};
 use lightning::util::ser::Writeable;
 
 use lightning_transaction_sync::EsploraSyncClient;
@@ -42,7 +42,7 @@ use bdk_esplora::EsploraAsyncExt;
 
 use esplora_client::AsyncClient as EsploraAsyncClient;
 
-use bitcoin::{FeeRate, Network};
+use bitcoin::{FeeRate, Network, Script, ScriptBuf, Txid};
 
 use std::collections::HashMap;
 use std::sync::{Arc, Mutex, RwLock};
@@ -128,6 +128,8 @@ pub(crate) enum ChainSource {
 		server_url: String,
 		sync_config: ElectrumSyncConfig,
 		electrum_runtime_client: RwLock<Option<ElectrumRuntimeClient>>,
+		registered_txs_cache: Mutex<Vec<(Txid, ScriptBuf)>>,
+		registered_outputs_cache: Mutex<Vec<WatchedOutput>>,
 		onchain_wallet: Arc<Wallet>,
 		onchain_wallet_sync_status: Mutex<WalletSyncStatus>,
 		lightning_wallet_sync_status: Mutex<WalletSyncStatus>,
@@ -190,12 +192,16 @@ impl ChainSource {
 		node_metrics: Arc<RwLock<NodeMetrics>>,
 	) -> Self {
 		let electrum_runtime_client = RwLock::new(None);
+		let registered_txs_cache = Mutex::new(Vec::new());
+		let registered_outputs_cache = Mutex::new(Vec::new());
 		let onchain_wallet_sync_status = Mutex::new(WalletSyncStatus::Completed);
 		let lightning_wallet_sync_status = Mutex::new(WalletSyncStatus::Completed);
 		Self::Electrum {
 			server_url,
 			sync_config,
 			electrum_runtime_client,
+			registered_txs_cache,
+			registered_outputs_cache,
 			onchain_wallet,
 			onchain_wallet_sync_status,
 			lightning_wallet_sync_status,
@@ -236,13 +242,36 @@ impl ChainSource {
 
 	pub(crate) fn start(&self, runtime: Arc<tokio::runtime::Runtime>) -> Result<(), Error> {
 		match self {
-			Self::Electrum { server_url, electrum_runtime_client, logger, .. } => {
+			Self::Electrum {
+				server_url,
+				electrum_runtime_client,
+				registered_txs_cache,
+				registered_outputs_cache,
+				logger,
+				..
+			} => {
 				let mut locked_client = electrum_runtime_client.write().unwrap();
-				*locked_client = Some(ElectrumRuntimeClient::new(
-					server_url.clone(),
-					runtime,
-					Arc::clone(&logger),
-				)?);
+				let client =
+					ElectrumRuntimeClient::new(server_url.clone(), runtime, Arc::clone(&logger))?;
+
+				// Apply any previously-cached `Filter` entries
+				{
+					let registered_txs_cache =
+						core::mem::take(&mut *registered_txs_cache.lock().unwrap());
+					for (txid, script_pubkey) in registered_txs_cache {
+						client.register_tx(&txid, &script_pubkey);
+					}
+				}
+
+				{
+					let registered_outputs_cache =
+						core::mem::take(&mut *registered_outputs_cache.lock().unwrap());
+					for output in registered_outputs_cache {
+						client.register_output(output)
+					}
+				}
+
+				*locked_client = Some(client);
 			},
 			_ => {
 				// Nothing to do for other chain sources.
@@ -1213,17 +1242,29 @@ impl ChainSource {
 }
 
 impl Filter for ChainSource {
-	fn register_tx(&self, txid: &bitcoin::Txid, script_pubkey: &bitcoin::Script) {
+	fn register_tx(&self, txid: &Txid, script_pubkey: &Script) {
 		match self {
 			Self::Esplora { tx_sync, .. } => tx_sync.register_tx(txid, script_pubkey),
-			Self::Electrum { .. } => todo!(),
+			Self::Electrum { electrum_runtime_client, registered_txs_cache, .. } => {
+				match electrum_runtime_client.read().unwrap().as_ref() {
+					Some(client) => client.register_tx(txid, script_pubkey),
+					None => {
+						registered_txs_cache.lock().unwrap().push((*txid, script_pubkey.to_owned()))
+					},
+				}
+			},
 			Self::BitcoindRpc { .. } => (),
 		}
 	}
 	fn register_output(&self, output: lightning::chain::WatchedOutput) {
 		match self {
 			Self::Esplora { tx_sync, .. } => tx_sync.register_output(output),
-			Self::Electrum { .. } => todo!(),
+			Self::Electrum { electrum_runtime_client, registered_outputs_cache, .. } => {
+				match electrum_runtime_client.read().unwrap().as_ref() {
+					Some(client) => client.register_output(output),
+					None => registered_outputs_cache.lock().unwrap().push(output),
+				}
+			},
 			Self::BitcoindRpc { .. } => (),
 		}
 	}