Retain Electrum Filterregistrations acrossstop/start``

src/chain/electrum.rs

@@ -5,7 +5,7 @@
 // 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 std::collections::HashMap;
+use std::collections::{HashMap, HashSet};
 use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::{Arc, Mutex, RwLock};
 use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
@@ -386,79 +386,78 @@ impl Filter for ElectrumChainSource {
 	}
 }
 
-enum ElectrumRuntimeStatus {
-	Started(Arc<ElectrumRuntimeClient>),
-	Stopped {
-		pending_registered_txs: Vec<(Txid, ScriptBuf)>,
-		pending_registered_outputs: Vec<WatchedOutput>,
-	},
+struct ElectrumRuntimeStatus {
+	client: Option<Arc<ElectrumRuntimeClient>>,
+	// The canonical inventory of all `Filter` entries registered over this chain source's
+	// lifetime.
+	//
+	// We retain these even while started: the `ElectrumRuntimeClient` (and hence the
+	// `ElectrumSyncClient` owning its own copy of the registrations) is dropped on `stop`, so
+	// replaying the inventory on the next `start` is the only way to keep watching the same set
+	// of transactions and outputs across a `stop`/`start` cycle. Note that we can't simply
+	// re-register from `ChannelMonitor::load_outputs_to_watch` instead, as, e.g., the
+	// `OutputSweeper` also registers outputs it wants to see spent.
+	registered_txs: HashMap<Txid, ScriptBuf>,
+	registered_outputs: HashSet<WatchedOutput>,
 }
 
 impl ElectrumRuntimeStatus {
 	fn new() -> Self {
-		let pending_registered_txs = Vec::new();
-		let pending_registered_outputs = Vec::new();
-		Self::Stopped { pending_registered_txs, pending_registered_outputs }
+		let client = None;
+		let registered_txs = HashMap::new();
+		let registered_outputs = HashSet::new();
+		Self { client, registered_txs, registered_outputs }
 	}
 
 	pub(super) fn start(
 		&mut self, server_url: String, sync_config: ElectrumSyncConfig, runtime: Arc<Runtime>,
 		config: Arc<Config>, logger: Arc<Logger>,
 	) -> Result<(), Error> {
-		match self {
-			Self::Stopped { pending_registered_txs, pending_registered_outputs } => {
-				let client = Arc::new(ElectrumRuntimeClient::new(
-					server_url,
-					sync_config,
-					runtime,
-					config,
-					logger,
-				)?);
-
-				// Apply any pending `Filter` entries
-				for (txid, script_pubkey) in pending_registered_txs.drain(..) {
-					client.register_tx(&txid, &script_pubkey);
-				}
+		if self.client.is_some() {
+			debug_assert!(false, "We shouldn't call start if we're already started");
+			return Ok(());
+		}
 
-				for output in pending_registered_outputs.drain(..) {
-					client.register_output(output)
-				}
+		let client =
+			Arc::new(ElectrumRuntimeClient::new(server_url, sync_config, runtime, config, logger)?);
 
-				*self = Self::Started(client);
-			},
-			Self::Started(_) => {
-				debug_assert!(false, "We shouldn't call start if we're already started")
-			},
+		// (Re-)apply all known `Filter` entries to the fresh client.
+		for (txid, script_pubkey) in self.registered_txs.iter() {
+			client.register_tx(txid, script_pubkey);
+		}
+
+		for output in self.registered_outputs.iter() {
+			client.register_output(output.clone());
 		}
+
+		self.client = Some(client);
+
 		Ok(())
 	}
 
 	pub(super) fn stop(&mut self) {
-		*self = Self::new()
+		// Drop the client, but retain the registration inventory so we can replay it if we're
+		// started again.
+		self.client = None;
 	}
 
 	fn client(&self) -> Option<Arc<ElectrumRuntimeClient>> {
-		match self {
-			Self::Started(client) => Some(Arc::clone(&client)),
-			Self::Stopped { .. } => None,
-		}
+		self.client.as_ref().map(Arc::clone)
 	}
 
 	fn register_tx(&mut self, txid: &Txid, script_pubkey: &Script) {
-		match self {
-			Self::Started(client) => client.register_tx(txid, script_pubkey),
-			Self::Stopped { pending_registered_txs, .. } => {
-				pending_registered_txs.push((*txid, script_pubkey.to_owned()))
-			},
+		self.registered_txs.insert(*txid, script_pubkey.to_owned());
+
+		if let Some(client) = self.client.as_ref() {
+			client.register_tx(txid, script_pubkey);
 		}
 	}
 
 	fn register_output(&mut self, output: lightning::chain::WatchedOutput) {
-		match self {
-			Self::Started(client) => client.register_output(output),
-			Self::Stopped { pending_registered_outputs, .. } => {
-				pending_registered_outputs.push(output)
-			},
+		self.registered_outputs.insert(output.clone());
+
+		if let Some(client) = self.client.as_ref() {
+			client.register_output(output);
 		}
 	}
 }

tests/integration_tests_rust.rs

@@ -24,10 +24,10 @@ use common::{
 	expect_channel_pending_event, expect_channel_ready_event, expect_channel_ready_events,
 	expect_event, expect_payment_claimable_event, expect_payment_received_event,
 	expect_payment_successful_event, expect_splice_negotiated_event, generate_blocks_and_wait,
-	generate_listening_addresses, invalidate_blocks, open_channel, open_channel_push_amt,
-	open_channel_with_all, premine_and_distribute_funds, premine_blocks, prepare_rbf,
-	random_chain_source, random_config, setup_bitcoind_and_electrsd, setup_builder, setup_node,
-	setup_two_nodes, splice_in_with_all, wait_for_block, wait_for_tx, InMemoryStore,
+	generate_listening_addresses, invalidate_blocks, open_channel, open_channel_no_wait,
+	open_channel_push_amt, open_channel_with_all, premine_and_distribute_funds, premine_blocks,
+	prepare_rbf, random_chain_source, random_config, setup_bitcoind_and_electrsd, setup_builder,
+	setup_node, setup_two_nodes, splice_in_with_all, wait_for_block, wait_for_tx, InMemoryStore,
 	TestChainSource, TestConfig, TestStoreType, TestSyncStore,
 };
 use electrsd::corepc_node::{self, Node as BitcoinD};
@@ -696,6 +696,61 @@ async fn start_stop_with_pathfinding_scores_sync() {
 	node.stop().unwrap();
 }
 
+// The Electrum chain source drops its runtime client - and with it the tx-sync client holding all
+// `Filter` registrations - when stopped. As `ChannelMonitor`s only register their watched
+// transactions and outputs while being loaded in `Builder::build`, nothing would re-register them
+// on the next `start`, leaving the node blind to confirmations and spends of, e.g., its funding
+// outputs. So here we assert the chain source replays its registrations across a `stop`/`start`
+// cycle.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn electrum_registrations_survive_chain_source_restart() {
+	let (bitcoind, electrsd) = setup_bitcoind_and_electrsd();
+	let chain_source = TestChainSource::Electrum(&electrsd);
+	let (node_a, node_b) = setup_two_nodes(&chain_source, false, false);
+
+	let address_a = node_a.onchain_payment().new_address().unwrap();
+
+	let premine_amount_sat = 5_000_000;
+	premine_and_distribute_funds(
+		&bitcoind.client,
+		&electrsd.client,
+		vec![address_a],
+		Amount::from_sat(premine_amount_sat),
+	)
+	.await;
+	node_a.sync_wallets().unwrap();
+
+	// Opening the channel registers the funding transaction and output with the chain source's
+	// `Filter`. We leave it unconfirmed for now, so watching for it is still pending.
+	let funding_txo = open_channel_no_wait(&node_a, &node_b, 4_000_000, None, false).await;
+	wait_for_tx(&electrsd.client, funding_txo.txid).await;
+
+	// Restart node A repeatedly, which tears down and recreates its Electrum chain source every time.
+	// Note that the `ChannelMonitor`s are not reloaded here, so the registrations have to survive in
+	// the chain source itself, and they have to survive more than a single cycle, i.e., replaying
+	// them mustn't consume them.
+	for _ in 0..3 {
+		node_a.stop().unwrap();
+		node_a.start().unwrap();
+	}
+
+	// Reconnect eagerly rather than waiting on the background reconnection interval.
+	let node_addr_b = node_b.listening_addresses().unwrap().first().unwrap().clone();
+	node_a.connect(node_b.node_id(), node_addr_b, false).unwrap();
+
+	generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 6).await;
+	node_a.sync_wallets().unwrap();
+	node_b.sync_wallets().unwrap();
+
+	// Node A can only learn that the funding transaction confirmed if its registrations survived
+	// the restart.
+	expect_channel_ready_event!(node_a, node_b.node_id());
+	expect_channel_ready_event!(node_b, node_a.node_id());
+
+	node_a.stop().unwrap();
+	node_b.stop().unwrap();
+}
+
 #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
 async fn onchain_send_receive() {
 	let (bitcoind, electrsd) = setup_bitcoind_and_electrsd();