From 3e40819b74513866298d09c9fd6bfc3b289dd180 Mon Sep 17 00:00:00 2001 From: Elias Rohrer Date: Tue, 18 Aug 2026 11:13:09 +0200 Subject: [PATCH] Prevent Electrum updates during shutdown A timed-out synchronous Electrum sync can outlive its driving task and continue calling Confirm implementations while node shutdown is draining persistence work. Close a gate before task cancellation so no new callbacks can start. Let callbacks already in progress finish atomically before shutdown continues. Co-Authored-By: HAL 9000 --- src/chain/electrum.rs | 263 +++++++++++++++++++++++++++++++++++++++++- src/chain/mod.rs | 12 ++ src/lib.rs | 4 + 3 files changed, 273 insertions(+), 6 deletions(-) diff --git a/src/chain/electrum.rs b/src/chain/electrum.rs index e266a1874..86025998e 100644 --- a/src/chain/electrum.rs +++ b/src/chain/electrum.rs @@ -7,7 +7,7 @@ use std::collections::{HashMap, HashSet}; use std::sync::atomic::{AtomicBool, Ordering}; -use std::sync::{Arc, Mutex, RwLock}; +use std::sync::{Arc, Mutex, RwLock, Weak}; use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; use bdk_chain::bdk_core::spk_client::{ @@ -95,7 +95,17 @@ impl ElectrumChainSource { } pub(super) fn stop(&self) { - self.electrum_runtime_status.write().expect("lock").stop(); + let client = self.electrum_runtime_status.write().expect("lock").stop(); + if let Some(client) = client { + client.begin_shutdown(); + } + } + + pub(super) fn begin_shutdown(&self) { + let client = self.electrum_runtime_status.read().expect("lock").client(); + if let Some(client) = client { + client.begin_shutdown(); + } } pub(crate) async fn sync_onchain_wallet( @@ -243,7 +253,7 @@ impl ElectrumChainSource { let sync_cman = Arc::clone(&channel_manager); let sync_cmon = Arc::clone(&chain_monitor); let sync_sweeper = Arc::clone(&output_sweeper); - let confirmables = vec![ + let confirmables: Vec> = vec![ sync_cman as Arc, sync_cmon as Arc, sync_sweeper as Arc, @@ -261,7 +271,8 @@ impl ElectrumChainSource { return Err(Error::TxSyncFailed); }; - let res = electrum_client.sync_confirmables(confirmables).await; + let confirmable = electrum_client.wrap_confirmables(&confirmables); + let res = electrum_client.sync_confirmables(vec![confirmable]).await; if let Ok(_) = res { let unix_time_secs_opt = @@ -436,10 +447,10 @@ impl ElectrumRuntimeStatus { Ok(()) } - pub(super) fn stop(&mut self) { + pub(super) fn stop(&mut self) -> Option> { // Drop the client, but retain the registration inventory so we can replay it if we're // started again. - self.client = None; + self.client.take() } fn client(&self) -> Option> { @@ -471,6 +482,7 @@ struct ElectrumRuntimeClient { runtime: Arc, config: Arc, logger: Arc, + confirm_gate: Arc, } impl ElectrumRuntimeClient { @@ -507,9 +519,23 @@ impl ElectrumRuntimeClient { runtime, config, logger, + confirm_gate: Arc::new(ConfirmGate::new()), }) } + fn begin_shutdown(&self) { + self.confirm_gate.deactivate(); + } + + fn wrap_confirmables( + &self, confirmables: &[Arc], + ) -> Arc { + Arc::new(ShutdownAwareConfirm::new( + Arc::downgrade(&self.confirm_gate), + confirmables.iter().map(Arc::downgrade).collect(), + )) + } + async fn sync_confirmables( &self, confirmables: Vec>, ) -> Result<(), Error> { @@ -539,6 +565,10 @@ impl ElectrumRuntimeClient { Error::TxSyncFailed })?; + if !self.confirm_gate.is_active() { + return Err(Error::TxSyncFailed); + } + log_debug!( self.logger, "Sync of Lightning wallet finished in {}ms.", @@ -795,6 +825,89 @@ impl ElectrumRuntimeClient { } } +struct ConfirmGate { + active: Mutex, +} + +impl ConfirmGate { + fn new() -> Self { + Self { active: Mutex::new(true) } + } + + fn deactivate(&self) { + *self.active.lock().expect("lock") = false; + } + + fn is_active(&self) -> bool { + *self.active.lock().expect("lock") + } +} + +struct ShutdownAwareConfirm { + gate: Weak, + confirmables: Vec>, +} + +impl ShutdownAwareConfirm { + fn new(gate: Weak, confirmables: Vec>) -> Self { + Self { gate, confirmables } + } + + fn with_confirmables( + &self, inactive_result: T, f: impl FnOnce(&[Arc]) -> T, + ) -> T { + let Some(gate) = self.gate.upgrade() else { + return inactive_result; + }; + let active = gate.active.lock().expect("lock"); + if !*active { + return inactive_result; + } + + let Some(confirmables): Option>> = + self.confirmables.iter().map(Weak::upgrade).collect() + else { + return inactive_result; + }; + f(&confirmables) + } +} + +impl Confirm for ShutdownAwareConfirm { + fn transactions_confirmed( + &self, header: &bitcoin::block::Header, + txdata: &lightning::chain::transaction::TransactionData<'_>, height: u32, + ) { + self.with_confirmables((), |confirmables| { + for confirmable in confirmables { + confirmable.transactions_confirmed(header, txdata, height); + } + }) + } + + fn transaction_unconfirmed(&self, txid: &Txid) { + self.with_confirmables((), |confirmables| { + for confirmable in confirmables { + confirmable.transaction_unconfirmed(txid); + } + }) + } + + fn best_block_updated(&self, header: &bitcoin::block::Header, height: u32) { + self.with_confirmables((), |confirmables| { + for confirmable in confirmables { + confirmable.best_block_updated(header, height); + } + }) + } + + fn get_relevant_txids(&self) -> Vec<(Txid, u32, Option)> { + self.with_confirmables(Vec::new(), |confirmables| { + confirmables.iter().flat_map(|confirmable| confirmable.get_relevant_txids()).collect() + }) + } +} + impl Filter for ElectrumRuntimeClient { fn register_tx(&self, txid: &Txid, script_pubkey: &Script) { self.tx_sync.register_tx(txid, script_pubkey) @@ -803,3 +916,141 @@ impl Filter for ElectrumRuntimeClient { self.tx_sync.register_output(output) } } + +#[cfg(test)] +mod tests { + use std::sync::atomic::{AtomicUsize, Ordering}; + use std::sync::mpsc; + use std::thread; + + use bitcoin::blockdata::constants::genesis_block; + + use super::*; + + struct RecordingConfirm { + calls: AtomicUsize, + relevant_txid: Txid, + } + + impl RecordingConfirm { + fn new(relevant_txid: Txid) -> Self { + Self { calls: AtomicUsize::new(0), relevant_txid } + } + } + + impl Confirm for RecordingConfirm { + fn transactions_confirmed( + &self, _header: &bitcoin::block::Header, + _txdata: &lightning::chain::transaction::TransactionData<'_>, _height: u32, + ) { + self.calls.fetch_add(1, Ordering::AcqRel); + } + + fn transaction_unconfirmed(&self, _txid: &Txid) { + self.calls.fetch_add(1, Ordering::AcqRel); + } + + fn best_block_updated(&self, _header: &bitcoin::block::Header, _height: u32) { + self.calls.fetch_add(1, Ordering::AcqRel); + } + + fn get_relevant_txids(&self) -> Vec<(Txid, u32, Option)> { + vec![(self.relevant_txid, 0, None)] + } + } + + struct BlockingConfirm { + calls: AtomicUsize, + started: Mutex>>, + release: Mutex>, + } + + impl Confirm for BlockingConfirm { + fn transactions_confirmed( + &self, _header: &bitcoin::block::Header, + _txdata: &lightning::chain::transaction::TransactionData<'_>, _height: u32, + ) { + } + + fn transaction_unconfirmed(&self, _txid: &Txid) {} + + fn best_block_updated(&self, _header: &bitcoin::block::Header, _height: u32) { + self.calls.fetch_add(1, Ordering::AcqRel); + if let Some(started) = self.started.lock().expect("lock").take() { + started.send(()).expect("test should still be waiting"); + } + self.release.lock().expect("lock").recv().expect("test should release callback"); + } + + fn get_relevant_txids(&self) -> Vec<(Txid, u32, Option)> { + Vec::new() + } + } + + #[test] + fn confirm_callbacks_are_ignored_after_shutdown() { + let block = genesis_block(Network::Regtest); + let txid = block.txdata[0].compute_txid(); + let delegate = Arc::new(RecordingConfirm::new(txid)); + let delegate_dyn: Arc = delegate.clone(); + let gate = Arc::new(ConfirmGate::new()); + let confirm = + ShutdownAwareConfirm::new(Arc::downgrade(&gate), vec![Arc::downgrade(&delegate_dyn)]); + + confirm.best_block_updated(&block.header, 0); + assert_eq!(delegate.calls.load(Ordering::Acquire), 1); + assert_eq!(confirm.get_relevant_txids(), vec![(txid, 0, None)]); + + gate.deactivate(); + confirm.transactions_confirmed(&block.header, &[], 0); + confirm.transaction_unconfirmed(&txid); + confirm.best_block_updated(&block.header, 0); + assert_eq!(delegate.calls.load(Ordering::Acquire), 1); + assert!(confirm.get_relevant_txids().is_empty()); + } + + #[test] + fn shutdown_waits_for_the_whole_confirm_callback() { + let block = genesis_block(Network::Regtest); + let txid = block.txdata[0].compute_txid(); + let (started_sender, started_receiver) = mpsc::sync_channel(1); + let (release_sender, release_receiver) = mpsc::sync_channel(1); + let blocking = Arc::new(BlockingConfirm { + calls: AtomicUsize::new(0), + started: Mutex::new(Some(started_sender)), + release: Mutex::new(release_receiver), + }); + let trailing = Arc::new(RecordingConfirm::new(txid)); + let blocking_dyn: Arc = blocking.clone(); + let trailing_dyn: Arc = trailing.clone(); + let gate = Arc::new(ConfirmGate::new()); + let confirm = Arc::new(ShutdownAwareConfirm::new( + Arc::downgrade(&gate), + vec![Arc::downgrade(&blocking_dyn), Arc::downgrade(&trailing_dyn)], + )); + + let callback = { + let confirm = Arc::clone(&confirm); + thread::spawn(move || confirm.best_block_updated(&block.header, 0)) + }; + started_receiver.recv().expect("callback should start"); + + let (shutdown_done_sender, shutdown_done_receiver) = mpsc::sync_channel(1); + let shutdown = thread::spawn(move || { + gate.deactivate(); + shutdown_done_sender.send(()).expect("test should still be waiting"); + }); + assert!(shutdown_done_receiver.recv_timeout(Duration::from_millis(50)).is_err()); + + release_sender.send(()).expect("callback should still be running"); + callback.join().expect("callback should finish"); + shutdown_done_receiver.recv().expect("shutdown should finish"); + shutdown.join().expect("shutdown should not panic"); + + assert_eq!(blocking.calls.load(Ordering::Acquire), 1); + assert_eq!(trailing.calls.load(Ordering::Acquire), 1); + confirm.best_block_updated(&block.header, 0); + assert_eq!(blocking.calls.load(Ordering::Acquire), 1); + assert_eq!(trailing.calls.load(Ordering::Acquire), 1); + } +} diff --git a/src/chain/mod.rs b/src/chain/mod.rs index 4aeb64704..ba7f798fa 100644 --- a/src/chain/mod.rs +++ b/src/chain/mod.rs @@ -255,6 +255,18 @@ impl ChainSource { } } + pub(crate) fn begin_shutdown(&self) { + match &self.kind { + ChainSourceKind::Electrum(electrum_chain_source) => { + electrum_chain_source.begin_shutdown() + }, + _ => { + // Other chain sources don't leave synchronous callbacks running after their + // driving future is cancelled. + }, + } + } + pub(crate) fn as_utxo_source(&self) -> Option { match &self.kind { ChainSourceKind::Bitcoind(bitcoind_chain_source) => { diff --git a/src/lib.rs b/src/lib.rs index 2bee539f7..dfc2b2641 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -845,6 +845,10 @@ impl Node { log_info!(self.logger, "Shutting down LDK Node with node ID {}...", self.node_id()); + // Prevent blocking Electrum syncs from making any further callbacks before persistence + // tasks stop accepting work. + self.chain_source.begin_shutdown(); + // Stop background tasks. self.stop_sender .send(())