From b9d645ad9cc8e180304afd2eb417e66ef19bf0e0 Mon Sep 17 00:00:00 2001 From: jolah1 Date: Thu, 6 Aug 2026 21:48:53 +0100 Subject: [PATCH 1/2] Retain Electrum `Filter` registrations across `stop`/`start` `ElectrumRuntimeStatus::stop` reset itself to `Stopped` with empty pending registration vectors, dropping the `ElectrumRuntimeClient` and, with it, the `ElectrumSyncClient` that owns the registered transactions and outputs. `start` then drained the pending vectors, so nothing survived even a single cycle. As `ChannelMonitor`s only register their watched transactions and outputs while being loaded in `Builder::build`, a `stop`/`start` cycle that doesn't rebuild the node left the chain source with no registrations at all, i.e., the node would no longer learn about confirmations or spends of, e.g., its funding outputs. Note we can't simply re-register from `ChannelMonitor::load_outputs_to_watch` on `start` either, as, e.g., the `OutputSweeper` also registers outputs it wants to see spent. Here we therefore keep a canonical, deduplicated registration inventory in `ElectrumRuntimeStatus` that is maintained whether we're started or not: `register_tx`/`register_output` always record the entry and additionally forward it to the client if one is live, `start` replays the inventory to the fresh client without consuming it, and `stop` merely drops the client. This also restores parity with the Esplora chain source, whose `tx_sync` is long-lived and hence never loses its registrations. Co-Authored-By: Claude Opus 5 --- src/chain/electrum.rs | 93 +++++++++++++++++++++---------------------- 1 file changed, 46 insertions(+), 47 deletions(-) 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); } } } From 880523311ccdfd493ebab6c23bb3b783f49b7bac Mon Sep 17 00:00:00 2001 From: jolah1 Date: Thu, 6 Aug 2026 21:49:07 +0100 Subject: [PATCH 2/2] Add test asserting Electrum registrations survive a restart Open a channel, leave the funding transaction unconfirmed, then repeatedly stop and start the node before confirming it. The node can only emit `ChannelReady` if its chain source replayed the `Filter` registrations on every `start`, so this fails without the preceding fix (the node times out waiting for the event) and passes with it. Co-Authored-By: Claude Opus 5 --- tests/integration_tests_rust.rs | 63 ++++++++++++++++++++++++++++++--- 1 file changed, 59 insertions(+), 4 deletions(-) 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();