diff --git a/src/chain/electrum.rs b/src/chain/electrum.rs index 59fa23a6c..5bae42665 100644 --- a/src/chain/electrum.rs +++ b/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), - Stopped { - pending_registered_txs: Vec<(Txid, ScriptBuf)>, - pending_registered_outputs: Vec, - }, +struct ElectrumRuntimeStatus { + client: Option>, + // 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, + registered_outputs: HashSet, } 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, config: Arc, logger: Arc, ) -> 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> { - 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); } } } diff --git a/tests/integration_tests_rust.rs b/tests/integration_tests_rust.rs index df477588f..2c36c2b15 100644 --- a/tests/integration_tests_rust.rs +++ b/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();