From 37b70b0e8745c2f654efc0f649a18283337400d3 Mon Sep 17 00:00:00 2001 From: Elias Rohrer Date: Tue, 13 Oct 2026 11:30:59 +0200 Subject: [PATCH 1/5] Separate initial sync from continuous polling Bitcoin Core listener initialization and continuous polling share one background task, making changes to their startup ordering harder to review independently. Extract initialization while keeping it in the same background task. Preserve retry, cancellation, and polling behavior so startup ordering can change separately. Co-Authored-By: HAL 9000 --- src/chain/bitcoind.rs | 33 +++++++++++++++++++++++++++------ 1 file changed, 27 insertions(+), 6 deletions(-) diff --git a/src/chain/bitcoind.rs b/src/chain/bitcoind.rs index 9ed38c2128..d4427c251a 100644 --- a/src/chain/bitcoind.rs +++ b/src/chain/bitcoind.rs @@ -178,11 +178,11 @@ impl BitcoindChainSource { Ok(()) } - pub(super) async fn continuously_sync_wallets( + pub(super) async fn initialize_wallets( &self, mut stop_sync_receiver: tokio::sync::watch::Receiver<()>, onchain_wallet: Arc, channel_manager: Arc, chain_monitor: Arc, output_sweeper: Arc, - ) { + ) -> Result<(), Error> { // First register for the wallet polling status to make sure `Node::sync_wallets` calls // wait on the result before proceeding. let Some(initial_sync_guard) = @@ -190,7 +190,7 @@ impl BitcoindChainSource { .await else { log_trace!(self.logger, "Stopping initial chain sync."); - return; + return Err(Error::TxSyncFailed); }; log_info!( @@ -205,7 +205,7 @@ impl BitcoindChainSource { // if the stop_sync_sender has been dropped, we should just exit if stop_sync_receiver.has_changed().unwrap_or(true) { log_trace!(self.logger, "Stopping initial chain sync."); - return; + return Err(Error::TxSyncFailed); } let onchain_wallet_best_block = onchain_wallet.current_best_block(); @@ -298,7 +298,7 @@ impl BitcoindChainSource { self.logger, "Stopping initial chain sync.", ); - return; + return Err(Error::TxSyncFailed); } _ = tokio::time::sleep(Duration::from_secs(backoff)) => {} } @@ -318,7 +318,7 @@ impl BitcoindChainSource { self.logger, "Stopping initial chain sync during backoff.", ); - return; + return Err(Error::TxSyncFailed); } _ = tokio::time::sleep(Duration::from_secs(MAX_BACKOFF_SECS)) => {} } @@ -329,6 +329,27 @@ impl BitcoindChainSource { // Now propagate the initial result to unblock waiting subscribers. initial_sync_guard.complete(Ok(())); + Ok(()) + } + + pub(super) async fn continuously_sync_wallets( + &self, mut stop_sync_receiver: tokio::sync::watch::Receiver<()>, + onchain_wallet: Arc, channel_manager: Arc, + chain_monitor: Arc, output_sweeper: Arc, + ) { + if self + .initialize_wallets( + stop_sync_receiver.clone(), + Arc::clone(&onchain_wallet), + Arc::clone(&channel_manager), + Arc::clone(&chain_monitor), + Arc::clone(&output_sweeper), + ) + .await + .is_err() + { + return; + } let mut chain_polling_interval = tokio::time::interval(Duration::from_secs(CHAIN_POLLING_INTERVAL_SECS)); From 875918de0a8af744b8e398b8c29272babeb83307 Mon Sep 17 00:00:00 2001 From: Elias Rohrer Date: Tue, 13 Oct 2026 12:36:18 +0200 Subject: [PATCH 2/5] Wait for chain sync before starting peers Bitcoin Core synchronized restored chain listeners in a background loop after startup. Peers and event processing could run while those listeners were still catching up, and sync failures only retried. Wait for initial synchronization before starting peers and event processing. Return synchronization errors to the caller so a failed start can shut down cleanly and be retried. Co-Authored-By: HAL 9000 --- src/chain/bitcoind.rs | 217 ++++++++++------------------ src/chain/bitcoind/startup_tests.rs | 55 +++++++ src/chain/mod.rs | 27 ++++ src/lib.rs | 10 ++ 4 files changed, 171 insertions(+), 138 deletions(-) create mode 100644 src/chain/bitcoind/startup_tests.rs diff --git a/src/chain/bitcoind.rs b/src/chain/bitcoind.rs index d4427c251a..394ca4593b 100644 --- a/src/chain/bitcoind.rs +++ b/src/chain/bitcoind.rs @@ -52,6 +52,9 @@ const CHAIN_POLLING_TIMEOUT_SECS: u64 = 10; type BitcoindSpvClient = SpvClient, BitcoindClient>, Arc>; +#[cfg(test)] +mod startup_tests; + async fn acquire_initial_wallet_sync_guard<'a>( wallet_polling_status: &'a Mutex, stop_sync_receiver: &mut tokio::sync::watch::Receiver<()>, @@ -198,133 +201,85 @@ impl BitcoindChainSource { "Starting initial synchronization of chain listeners. This might take a while..", ); - let mut backoff = CHAIN_POLLING_INTERVAL_SECS; - const MAX_BACKOFF_SECS: u64 = 300; - - loop { - // if the stop_sync_sender has been dropped, we should just exit - if stop_sync_receiver.has_changed().unwrap_or(true) { - log_trace!(self.logger, "Stopping initial chain sync."); - return Err(Error::TxSyncFailed); - } + // if the stop_sync_sender has been dropped, we should just exit + if stop_sync_receiver.has_changed().unwrap_or(true) { + log_trace!(self.logger, "Stopping initial chain sync."); + return Err(Error::TxSyncFailed); + } - let onchain_wallet_best_block = onchain_wallet.current_best_block(); - let channel_manager_best_block = channel_manager.current_best_block(); - let sweeper_best_block = output_sweeper.current_best_block(); - - let mut chain_listeners = vec![ - (onchain_wallet_best_block, &*onchain_wallet as &(dyn Listen + Send + Sync)), - (channel_manager_best_block, &*channel_manager as &(dyn Listen + Send + Sync)), - (sweeper_best_block, &*output_sweeper as &(dyn Listen + Send + Sync)), - ]; - - // TODO: Eventually we might want to see if we can synchronize `ChannelMonitor`s - // before giving them to `ChainMonitor` it the first place. However, this isn't - // trivial as we load them on initialization (in the `Builder`) and only gain - // network access during `start`. For now, we just make sure we get the worst known - // block hash and sychronize them via `ChainMonitor`. - if let Some(worst_channel_monitor_best_block) = chain_monitor - .list_monitors() - .iter() - .flat_map(|channel_id| chain_monitor.get_monitor(*channel_id)) - .map(|m| m.current_best_block()) - .min_by_key(|b| b.height) - { - chain_listeners.push(( - worst_channel_monitor_best_block, - &*chain_monitor as &(dyn Listen + Send + Sync), - )); - } + let onchain_wallet_best_block = onchain_wallet.current_best_block(); + let channel_manager_best_block = channel_manager.current_best_block(); + let sweeper_best_block = output_sweeper.current_best_block(); + + let mut chain_listeners = vec![ + (onchain_wallet_best_block, &*onchain_wallet as &(dyn Listen + Send + Sync)), + (channel_manager_best_block, &*channel_manager as &(dyn Listen + Send + Sync)), + (sweeper_best_block, &*output_sweeper as &(dyn Listen + Send + Sync)), + ]; + + // TODO: Eventually we might want to see if we can synchronize `ChannelMonitor`s + // before giving them to `ChainMonitor` it the first place. However, this isn't + // trivial as we load them on initialization (in the `Builder`) and only gain + // network access during `start`. For now, we just make sure we get the worst known + // block hash and sychronize them via `ChainMonitor`. + if let Some(worst_channel_monitor_best_block) = chain_monitor + .list_monitors() + .iter() + .flat_map(|channel_id| chain_monitor.get_monitor(*channel_id)) + .map(|m| m.current_best_block()) + .min_by_key(|b| b.height) + { + chain_listeners.push(( + worst_channel_monitor_best_block, + &*chain_monitor as &(dyn Listen + Send + Sync), + )); + } - let now = SystemTime::now(); - match synchronize_listeners( - self.api_client.as_ref(), - self.config.network, - chain_listeners.clone(), - ) - .await - { - Ok((header_cache, chain_tip)) => { - let spv_client = self.new_spv_client( - chain_tip, - header_cache, - Arc::clone(&onchain_wallet), - Arc::clone(&channel_manager), - Arc::clone(&chain_monitor), - Arc::clone(&output_sweeper), - ); - *self.spv_client.lock().await = Some(spv_client); - { - let elapsed_ms = now.elapsed().map(|d| d.as_millis()).unwrap_or(0); - log_info!( - self.logger, - "Finished synchronizing listeners in {}ms", - elapsed_ms, - ); - *self.latest_chain_tip.write().expect("lock") = Some(chain_tip); - let unix_time_secs_opt = - SystemTime::now().duration_since(UNIX_EPOCH).ok().map(|d| d.as_secs()); - update_and_persist_node_metrics( - &self.node_metrics, - &*self.kv_store, - &*self.logger, - |m| { - m.latest_lightning_wallet_sync_timestamp = unix_time_secs_opt; - m.latest_onchain_wallet_sync_timestamp = unix_time_secs_opt; - }, - ) - .await - .unwrap_or_else(|e| { - log_error!(self.logger, "Failed to persist node metrics: {}", e); - }); - } - break; - }, + let now = SystemTime::now(); + match synchronize_listeners( + self.api_client.as_ref(), + self.config.network, + chain_listeners.clone(), + ) + .await + { + Ok((header_cache, chain_tip)) => { + let spv_client = self.new_spv_client( + chain_tip, + header_cache, + Arc::clone(&onchain_wallet), + Arc::clone(&channel_manager), + Arc::clone(&chain_monitor), + Arc::clone(&output_sweeper), + ); + *self.spv_client.lock().await = Some(spv_client); + { + let elapsed_ms = now.elapsed().map(|d| d.as_millis()).unwrap_or(0); + log_info!(self.logger, "Finished synchronizing listeners in {}ms", elapsed_ms,); + *self.latest_chain_tip.write().expect("lock") = Some(chain_tip); + let unix_time_secs_opt = + SystemTime::now().duration_since(UNIX_EPOCH).ok().map(|d| d.as_secs()); + update_and_persist_node_metrics( + &self.node_metrics, + &*self.kv_store, + &*self.logger, + |m| { + m.latest_lightning_wallet_sync_timestamp = unix_time_secs_opt; + m.latest_onchain_wallet_sync_timestamp = unix_time_secs_opt; + }, + ) + .await + .unwrap_or_else(|e| { + log_error!(self.logger, "Failed to persist node metrics: {}", e); + }); + } + }, - Err(e) => { - log_error!(self.logger, "Failed to synchronize chain listeners: {:?}", e); - if e.kind() == BlockSourceErrorKind::Transient { - log_info!( - self.logger, - "Transient error syncing chain listeners: {:?}. Retrying in {} seconds.", - e, - backoff - ); - // Sleep with stop signal check to allow immediate shutdown - tokio::select! { - biased; - _ = stop_sync_receiver.changed() => { - log_trace!( - self.logger, - "Stopping initial chain sync.", - ); - return Err(Error::TxSyncFailed); - } - _ = tokio::time::sleep(Duration::from_secs(backoff)) => {} - } - backoff = std::cmp::min(backoff * 2, MAX_BACKOFF_SECS); - } else { - log_error!( - self.logger, - "Persistent error syncing chain listeners: {:?}. Retrying in {} seconds.", - e, - MAX_BACKOFF_SECS - ); - // Sleep with stop signal check to allow immediate shutdown - tokio::select! { - biased; - _ = stop_sync_receiver.changed() => { - log_trace!( - self.logger, - "Stopping initial chain sync during backoff.", - ); - return Err(Error::TxSyncFailed); - } - _ = tokio::time::sleep(Duration::from_secs(MAX_BACKOFF_SECS)) => {} - } - } - }, - } + Err(e) => { + log_error!(self.logger, "Failed to synchronize chain listeners: {:?}", e); + initial_sync_guard.complete(Err(Error::TxSyncFailed)); + return Err(Error::TxSyncFailed); + }, } // Now propagate the initial result to unblock waiting subscribers. @@ -337,20 +292,6 @@ impl BitcoindChainSource { onchain_wallet: Arc, channel_manager: Arc, chain_monitor: Arc, output_sweeper: Arc, ) { - if self - .initialize_wallets( - stop_sync_receiver.clone(), - Arc::clone(&onchain_wallet), - Arc::clone(&channel_manager), - Arc::clone(&chain_monitor), - Arc::clone(&output_sweeper), - ) - .await - .is_err() - { - return; - } - let mut chain_polling_interval = tokio::time::interval(Duration::from_secs(CHAIN_POLLING_INTERVAL_SECS)); chain_polling_interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); diff --git a/src/chain/bitcoind/startup_tests.rs b/src/chain/bitcoind/startup_tests.rs new file mode 100644 index 0000000000..65c5a93575 --- /dev/null +++ b/src/chain/bitcoind/startup_tests.rs @@ -0,0 +1,55 @@ +// This file is Copyright its original authors, visible in version control history. +// +// This file is licensed under the Apache License, Version 2.0 or the MIT license , at your option. You may not use this file except in +// accordance with one or both of these licenses. + +// The builder's Esplora backend lets us construct a node without network access. All +// synchronization in this test uses the Bitcoin Core startup path and the source below. +#![cfg(feature = "chain-esplora")] + +use std::sync::Arc; + +use bitcoin::Network; + +use crate::builder::NodeBuilder; +use crate::entropy::NodeEntropy; +use crate::io::test_utils::InMemoryStore; +use crate::types::DynStoreRef; +use crate::{Config, Error}; + +#[test] +fn startup_fails_if_bitcoind_sync_fails() { + let mut builder = + NodeBuilder::from_config(Config { network: Network::Testnet, ..Config::default() }); + builder.set_log_facade_logger(); + #[cfg(not(feature = "uniffi"))] + let entropy = NodeEntropy::from_seed_bytes([42; 64]); + #[cfg(feature = "uniffi")] + let entropy = NodeEntropy::from_seed_bytes(vec![42; 64]).unwrap(); + + // Persist a wallet first so the Core-backed build does not require fetching a + // birthday from the unavailable backend. This leaves startup responsible for sync. + let initialized_node = builder.build_with_store(entropy, InMemoryStore::new()).unwrap(); + let store = DynStoreRef(Arc::clone(&initialized_node.kv_store)); + drop(initialized_node); + builder.set_chain_source_bitcoind_rpc( + "127.0.0.1".to_owned(), + 0, + "test".to_owned(), + "test".to_owned(), + None, + ); + let node = builder.build_with_store(entropy, store).unwrap(); + + // Testnet tolerates unavailable fee estimates, so this failure must come from + // initial chain synchronization. Port zero cannot have a listening RPC server. + assert_eq!( + node.start(), + Err(Error::TxSyncFailed), + "startup must fail when initial Bitcoin Core synchronization fails", + ); + assert!(!node.status().is_running); + assert!(node.list_peers().is_empty()); +} diff --git a/src/chain/mod.rs b/src/chain/mod.rs index f01c1c8cb8..e69a071108 100644 --- a/src/chain/mod.rs +++ b/src/chain/mod.rs @@ -309,6 +309,33 @@ impl ChainSource { } } + pub(crate) async fn initialize_wallets( + &self, stop_sync_receiver: tokio::sync::watch::Receiver<()>, onchain_wallet: Arc, + channel_manager: Arc, chain_monitor: Arc, + output_sweeper: Arc, + ) -> Result<(), Error> { + #[cfg(not(feature = "chain-bitcoind"))] + let _ = (stop_sync_receiver, onchain_wallet, channel_manager, chain_monitor, output_sweeper); + match &self.kind { + #[cfg(feature = "chain-bitcoind")] + ChainSourceKind::Bitcoind(source) => { + source + .initialize_wallets( + stop_sync_receiver, + onchain_wallet, + channel_manager, + chain_monitor, + output_sweeper, + ) + .await + }, + #[cfg(feature = "chain-esplora")] + ChainSourceKind::Esplora(_) => Ok(()), + #[cfg(feature = "chain-electrum")] + ChainSourceKind::Electrum(_) => Ok(()), + } + } + pub(crate) async fn continuously_sync_wallets( &self, stop_sync_receiver: tokio::sync::watch::Receiver<()>, onchain_wallet: Arc, channel_manager: Arc, chain_monitor: Arc, diff --git a/src/lib.rs b/src/lib.rs index b1f98624f8..876257bfc6 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -369,6 +369,16 @@ impl Node { ) })?; + // Reconcile block-based listeners before peers or event processing can act on stale + // confirmations. Each restored ChannelMonitor may have been persisted at a different tip. + self.runtime.block_on(self.chain_source.initialize_wallets( + self.stop_sender.subscribe(), + Arc::clone(&self.wallet), + Arc::clone(&self.channel_manager), + Arc::clone(&self.chain_monitor), + Arc::clone(&self.output_sweeper), + ))?; + // Spawn background task continuously syncing onchain, lightning, and fee rate cache. let stop_sync_receiver = self.stop_sender.subscribe(); let chain_source = Arc::clone(&self.chain_source); From c4fa9de18f1bc8de201793c3aea18e764f10ba76 Mon Sep 17 00:00:00 2001 From: Elias Rohrer Date: Tue, 13 Oct 2026 12:39:25 +0200 Subject: [PATCH 3/5] Reconcile restored monitors from their own tips Bitcoin Core startup synchronized all restored channel monitors from the lowest saved tip. A monitor saved farther ahead on an orphaned fork could miss disconnects and retain stale confirmations. Reconcile each monitor from its own current locator while peers and event processing remain stopped. Cover divergent persisted tips and retries after partial synchronization with regression tests. Fixes #1076 Reported-by: Project Loupe / instagibbs Co-Authored-By: HAL 9000 --- src/chain/bitcoind.rs | 106 +++++++++--- src/chain/bitcoind/startup_tests.rs | 257 +++++++++++++++++++++++++++- src/chain/mod.rs | 1 + 3 files changed, 340 insertions(+), 24 deletions(-) diff --git a/src/chain/bitcoind.rs b/src/chain/bitcoind.rs index 394ca4593b..79edbe43a3 100644 --- a/src/chain/bitcoind.rs +++ b/src/chain/bitcoind.rs @@ -17,7 +17,9 @@ use base64::Engine; use bitcoin::transaction::Version; use bitcoin::{BlockHash, FeeRate, Network, OutPoint, Transaction, Txid}; use lightning::chain::chaininterface::ConfirmationTarget as LdkConfirmationTarget; +use lightning::chain::transaction::TransactionData; use lightning::chain::{BlockLocator, Listen}; +use lightning::ln::types::ChannelId; use lightning::util::ser::Writeable; use lightning_block_sync::gossip::UtxoSource; use lightning_block_sync::http::{HttpClientError, JsonResponse}; @@ -26,8 +28,8 @@ use lightning_block_sync::poll::{ChainPoller, ChainTip, ValidatedBlockHeader}; use lightning_block_sync::rest::RestClient; use lightning_block_sync::rpc::{RpcClient, RpcClientError}; use lightning_block_sync::{ - BlockData, BlockHeaderData, BlockSource, BlockSourceError, BlockSourceErrorKind, HeaderCache, - SpvClient, + BlockData, BlockHeaderData, BlockSource, BlockSourceError, BlockSourceErrorKind, + BlockSourceResult, HeaderCache, SpvClient, }; use serde::Serialize; @@ -43,7 +45,7 @@ use crate::fee_estimator::{ use crate::io::utils::update_and_persist_node_metrics; use crate::logger::{log_bytes, log_debug, log_error, log_info, log_trace, LdkLogger, Logger}; use crate::tx_broadcaster::SortedTransactions; -use crate::types::{ChainMonitor, ChannelManager, DynStore, Sweeper, Wallet}; +use crate::types::{Broadcaster, ChainMonitor, ChannelManager, DynStore, Sweeper, Wallet}; use crate::{Error, PersistedNodeMetrics}; const CHAIN_POLLING_INTERVAL_SECS: u64 = 2; @@ -55,6 +57,77 @@ type BitcoindSpvClient = #[cfg(test)] mod startup_tests; +/// Replays chain data to one restored monitor before aggregate polling begins. +struct ChannelMonitorListener<'a> { + channel_id: ChannelId, + chain_monitor: &'a ChainMonitor, + tx_broadcaster: &'a Broadcaster, + fee_estimator: &'a OnchainFeeEstimator, + logger: &'a Logger, +} + +impl Listen for ChannelMonitorListener<'_> { + fn filtered_block_connected( + &self, header: &bitcoin::block::Header, txdata: &TransactionData, height: u32, + ) { + let monitor = self.chain_monitor.get_monitor(self.channel_id).expect("loaded monitor"); + // Bitcoin Core supplies full blocks, so newly watched outputs need no registration. + monitor.block_connected( + header, + txdata, + height, + self.tx_broadcaster, + self.fee_estimator, + self.logger, + ); + } + + fn blocks_disconnected(&self, fork_point: BlockLocator) { + let monitor = self.chain_monitor.get_monitor(self.channel_id).expect("loaded monitor"); + monitor.blocks_disconnected( + fork_point, + self.tx_broadcaster, + self.fee_estimator, + self.logger, + ); + } +} + +async fn synchronize_chain_listeners( + block_source: &B, network: Network, chain_monitor: &ChainMonitor, tx_broadcaster: &Broadcaster, + fee_estimator: &OnchainFeeEstimator, logger: &Logger, + chain_listeners: Vec<(BlockLocator, &(dyn Listen + Send + Sync))>, +) -> BlockSourceResult<(HeaderCache, ValidatedBlockHeader)> { + // A single checkpoint for the aggregate ChainMonitor would miss disconnects for monitors + // persisted farther ahead on a stale fork. Pair every monitor with its own current locator, + // including on retries after a partially completed synchronization. + let monitor_listeners: Vec<_> = chain_monitor + .list_monitors() + .into_iter() + .map(|channel_id| ChannelMonitorListener { + channel_id, + chain_monitor, + tx_broadcaster, + fee_estimator, + logger, + }) + .collect(); + let mut listeners: Vec<_> = monitor_listeners + .iter() + .map(|listener| { + let best_block = chain_monitor + .get_monitor(listener.channel_id) + .expect("loaded monitor") + .current_best_block(); + (best_block, listener as &(dyn Listen + Send + Sync)) + }) + .collect(); + listeners.extend(chain_listeners); + // Replay need not be persisted here: it is repeated on restart, as permitted by + // ChainMonitor::load_existing_monitor. Peers and event processing must remain stopped. + synchronize_listeners(block_source, network, listeners).await +} + async fn acquire_initial_wallet_sync_guard<'a>( wallet_polling_status: &'a Mutex, stop_sync_receiver: &mut tokio::sync::watch::Receiver<()>, @@ -185,6 +258,7 @@ impl BitcoindChainSource { &self, mut stop_sync_receiver: tokio::sync::watch::Receiver<()>, onchain_wallet: Arc, channel_manager: Arc, chain_monitor: Arc, output_sweeper: Arc, + tx_broadcaster: &Broadcaster, ) -> Result<(), Error> { // First register for the wallet polling status to make sure `Node::sync_wallets` calls // wait on the result before proceeding. @@ -211,34 +285,20 @@ impl BitcoindChainSource { let channel_manager_best_block = channel_manager.current_best_block(); let sweeper_best_block = output_sweeper.current_best_block(); - let mut chain_listeners = vec![ + let chain_listeners = vec![ (onchain_wallet_best_block, &*onchain_wallet as &(dyn Listen + Send + Sync)), (channel_manager_best_block, &*channel_manager as &(dyn Listen + Send + Sync)), (sweeper_best_block, &*output_sweeper as &(dyn Listen + Send + Sync)), ]; - // TODO: Eventually we might want to see if we can synchronize `ChannelMonitor`s - // before giving them to `ChainMonitor` it the first place. However, this isn't - // trivial as we load them on initialization (in the `Builder`) and only gain - // network access during `start`. For now, we just make sure we get the worst known - // block hash and sychronize them via `ChainMonitor`. - if let Some(worst_channel_monitor_best_block) = chain_monitor - .list_monitors() - .iter() - .flat_map(|channel_id| chain_monitor.get_monitor(*channel_id)) - .map(|m| m.current_best_block()) - .min_by_key(|b| b.height) - { - chain_listeners.push(( - worst_channel_monitor_best_block, - &*chain_monitor as &(dyn Listen + Send + Sync), - )); - } - let now = SystemTime::now(); - match synchronize_listeners( + match synchronize_chain_listeners( self.api_client.as_ref(), self.config.network, + &chain_monitor, + tx_broadcaster, + &self.fee_estimator, + &self.logger, chain_listeners.clone(), ) .await diff --git a/src/chain/bitcoind/startup_tests.rs b/src/chain/bitcoind/startup_tests.rs index 65c5a93575..6526c6137e 100644 --- a/src/chain/bitcoind/startup_tests.rs +++ b/src/chain/bitcoind/startup_tests.rs @@ -9,16 +9,118 @@ // synchronization in this test uses the Bitcoin Core startup path and the source below. #![cfg(feature = "chain-esplora")] +use std::collections::HashMap; use std::sync::Arc; -use bitcoin::Network; +use bitcoin::block::{Header, Version}; +use bitcoin::constants::genesis_block; +use bitcoin::hashes::Hash; +use bitcoin::{Block, BlockHash, CompactTarget, Network, Transaction, TxMerkleNode}; +use lightning::chain::channelmonitor::ChannelMonitor; +use lightning::chain::{BlockLocator, Listen}; +use lightning::ln::functional_test_utils::{ + create_chan_between_nodes_with_value_init, create_chanmon_cfgs, create_network, + create_node_cfgs, create_node_chanmgrs, test_legacy_channel_config, +}; +use lightning::sign::{InMemorySigner, KeysManager}; +use lightning::util::ser::{ReadableArgs, Writeable}; +use lightning_block_sync::{ + BlockData, BlockHeaderData, BlockSource, BlockSourceError, BlockSourceErrorKind, + BlockSourceResult, +}; +use super::synchronize_chain_listeners; use crate::builder::NodeBuilder; use crate::entropy::NodeEntropy; use crate::io::test_utils::InMemoryStore; use crate::types::DynStoreRef; use crate::{Config, Error}; +struct ForkedBlockSource { + headers: HashMap, + best_block: BlockHash, + fail_block_fetch: bool, +} + +impl ForkedBlockSource { + fn new(genesis: &Block) -> Self { + let best_block = genesis.block_hash(); + let header = + BlockHeaderData { header: genesis.header, height: 0, chainwork: genesis.header.work() }; + Self { headers: HashMap::from([(best_block, header)]), best_block, fail_block_fetch: false } + } + + fn add_block(&mut self, block: &Block) { + let parent = self.headers.get(&block.header.prev_blockhash).unwrap(); + let header = BlockHeaderData { + header: block.header, + height: parent.height + 1, + chainwork: parent.chainwork + block.header.work(), + }; + self.headers.insert(block.block_hash(), header); + } +} + +impl BlockSource for ForkedBlockSource { + async fn get_header<'a>( + &'a self, block_hash: &'a BlockHash, _height_hint: Option, + ) -> BlockSourceResult { + self.headers + .get(block_hash) + .cloned() + .ok_or_else(|| BlockSourceError::persistent("unknown test block")) + } + + async fn get_block<'a>(&'a self, block_hash: &'a BlockHash) -> BlockSourceResult { + if self.fail_block_fetch { + return Err(BlockSourceError::transient("injected block fetch failure")); + } + // Only the empty replacement block is connected during synchronization. The + // original funding and commitment transactions have already been persisted. + assert_eq!(*block_hash, self.best_block); + Ok(BlockData::HeaderOnly(self.get_header(block_hash, None).await?.header)) + } + + async fn get_best_block(&self) -> BlockSourceResult<(BlockHash, Option)> { + Ok((self.best_block, Some(self.headers[&self.best_block].height))) + } +} + +fn next_block(parent: &Block, txdata: Vec, time_offset: u32) -> Block { + let mut block = Block { + header: Header { + version: Version::NO_SOFT_FORK_SIGNALLING, + prev_blockhash: parent.block_hash(), + merkle_root: TxMerkleNode::all_zeros(), + time: parent.header.time + time_offset, + bits: CompactTarget::from_consensus(0x207fffff), + nonce: 0, + }, + txdata, + }; + if let Some(merkle_root) = block.compute_merkle_root() { + block.header.merkle_root = merkle_root; + } + while block.header.validate_pow(block.header.target()).is_err() { + block.header.nonce += 1; + } + block +} + +fn read_monitor(bytes: &[u8], keys: &KeysManager) -> ChannelMonitor { + <(BlockLocator, ChannelMonitor)>::read(&mut &bytes[..], (keys, keys)).unwrap().1 +} + +#[test] +fn startup_reconciles_divergent_monitor_tips() { + do_test_startup_reconciles_divergent_monitor_tips(false); +} + +#[test] +fn startup_retries_after_disconnecting_monitor_blocks() { + do_test_startup_reconciles_divergent_monitor_tips(true); +} + #[test] fn startup_fails_if_bitcoind_sync_fails() { let mut builder = @@ -53,3 +155,156 @@ fn startup_fails_if_bitcoind_sync_fails() { assert!(!node.status().is_running); assert!(node.list_peers().is_empty()); } + +fn do_test_startup_reconciles_divergent_monitor_tips(fail_first_sync: bool) { + let network = Network::Testnet; + let genesis = genesis_block(network); + // LDK's node-0 fixture uses this seed and remote-key derivation mode. Deserializing + // its monitors with the underlying signer gives us the real Node monitor type. + let keys = KeysManager::new(&[0; 32], genesis.header.time as u64, 0, true); + let (monitors, funding_transactions) = { + let monitor_configs = create_chanmon_cfgs(3); + let node_configs = create_node_cfgs(3, &monitor_configs); + let channel_configs = vec![Some(test_legacy_channel_config()); 3]; + let channel_managers = create_node_chanmgrs(3, &node_configs, &channel_configs); + let nodes = create_network(3, &node_configs, &channel_managers); + let funding_transactions = vec![ + create_chan_between_nodes_with_value_init(&nodes[0], &nodes[1], 100_000, 10_001), + create_chan_between_nodes_with_value_init(&nodes[0], &nodes[2], 100_000, 10_001), + ]; + let chain_monitor = &nodes[0].chain_monitor.chain_monitor; + let monitors = chain_monitor + .list_monitors() + .into_iter() + .map(|channel_id| { + read_monitor(&chain_monitor.get_monitor(channel_id).unwrap().encode(), &keys) + }) + .collect::>(); + assert_eq!(monitors.len(), 2); + (monitors, funding_transactions) + }; + + let monitor_configs = create_chanmon_cfgs(1); + let monitor_config = &monitor_configs[0]; + let funding_block = next_block(&genesis, funding_transactions, 1); + for monitor in &monitors { + monitor.block_connected( + &funding_block.header, + &funding_block.txdata.iter().enumerate().collect::>(), + 1, + &monitor_config.tx_broadcaster, + &monitor_config.fee_estimator, + &monitor_config.logger, + ); + } + let commitment = + monitors[0].unsafe_get_latest_holder_commitment_txn(&monitor_config.logger).remove(0); + let commitment_txid = commitment.compute_txid(); + let orphan = next_block(&funding_block, vec![commitment], 1); + monitors[0].block_connected( + &orphan.header, + &orphan.txdata.iter().enumerate().collect::>(), + 2, + &monitor_config.tx_broadcaster, + &monitor_config.fee_estimator, + &monitor_config.logger, + ); + let ahead_channel_id = monitors[0].channel_id(); + let ahead_funding_txid = monitors[0].get_funding_txo().txid; + assert!(monitors[0].get_relevant_txids().contains(&( + commitment_txid, + 2, + Some(orphan.block_hash()), + ))); + assert_eq!(monitors[0].current_best_block().height, 2); + assert_eq!(monitors[1].current_best_block().height, 1); + + let mut builder = NodeBuilder::from_config(Config { network, ..Config::default() }); + builder.set_log_facade_logger(); + #[cfg(not(feature = "uniffi"))] + let entropy = NodeEntropy::from_seed_bytes([42; 64]); + #[cfg(feature = "uniffi")] + let entropy = NodeEntropy::from_seed_bytes(vec![42; 64]).unwrap(); + let node = builder.build_with_store(entropy, InMemoryStore::new()).unwrap(); + // Round-trip each persisted monitor at its own tip, just as after a restart. + for monitor in monitors { + let restored_monitor = read_monitor(&monitor.encode(), &keys); + node.chain_monitor + .load_existing_monitor(restored_monitor.channel_id(), restored_monitor) + .unwrap(); + } + + let replacement = next_block(&funding_block, Vec::new(), 2); + let mut source = ForkedBlockSource::new(&genesis); + source.add_block(&funding_block); + source.add_block(&orphan); + source.add_block(&replacement); + source.best_block = replacement.block_hash(); + let synchronize = |source: &ForkedBlockSource| { + node.runtime.block_on(synchronize_chain_listeners( + source, + network, + &node.chain_monitor, + &node.tx_broadcaster, + &node.fee_estimator, + &node.logger, + Vec::new(), + )) + }; + if fail_first_sync { + // All stale blocks are disconnected before replacement blocks are fetched. A + // fetch failure must therefore leave a partially synchronized monitor whose + // new locator is used on the next attempt. + source.fail_block_fetch = true; + let error = synchronize(&source).err().expect("block fetch must fail"); + assert_eq!(error.kind(), BlockSourceErrorKind::Transient); + let monitor = node.chain_monitor.get_monitor(ahead_channel_id).unwrap(); + let best_block = monitor.current_best_block(); + assert_eq!( + best_block.block_hash, + funding_block.block_hash(), + "failed synchronization must leave the monitor rewound to the common block", + ); + assert_eq!(best_block.height, 1); + assert!( + !monitor.get_relevant_txids().iter().any(|(txid, _, _)| *txid == commitment_txid), + "disconnects must remove stale confirmations before a block fetch can fail", + ); + source.fail_block_fetch = false; + } + synchronize(&source).unwrap(); + + let monitor = node.chain_monitor.get_monitor(ahead_channel_id).unwrap(); + let relevant_txids = monitor.get_relevant_txids(); + assert!( + !relevant_txids.iter().any(|(txid, _, _)| *txid == commitment_txid), + "startup must disconnect the orphaned commitment confirmation", + ); + assert!( + relevant_txids.contains(&(ahead_funding_txid, 1, Some(funding_block.block_hash()))), + "startup must preserve the funding confirmation on the common chain", + ); + drop(monitor); + for channel_id in node.chain_monitor.list_monitors() { + let best_block = node.chain_monitor.get_monitor(channel_id).unwrap().current_best_block(); + assert_eq!(best_block.block_hash, replacement.block_hash()); + assert_eq!(best_block.height, 2); + } + + // Once startup has reconciled the monitors, ordinary aggregate notifications must + // accept the same commitment confirming on the replacement chain. + let reconfirmation = next_block(&replacement, orphan.txdata.clone(), 1); + node.chain_monitor.filtered_block_connected( + &reconfirmation.header, + &reconfirmation.txdata.iter().enumerate().collect::>(), + 3, + ); + assert!( + node.chain_monitor.get_monitor(ahead_channel_id).unwrap().get_relevant_txids().contains(&( + commitment_txid, + 3, + Some(reconfirmation.block_hash()), + )), + "aggregate synchronization must track the commitment's new confirmation", + ); +} diff --git a/src/chain/mod.rs b/src/chain/mod.rs index e69a071108..7e1c908bed 100644 --- a/src/chain/mod.rs +++ b/src/chain/mod.rs @@ -326,6 +326,7 @@ impl ChainSource { channel_manager, chain_monitor, output_sweeper, + &self.tx_broadcaster, ) .await }, From f552c4b572b2524618678dd7abb157bc5e5b7004 Mon Sep 17 00:00:00 2001 From: Elias Rohrer Date: Tue, 13 Oct 2026 11:38:41 +0200 Subject: [PATCH 4/5] Sync Lightning state before all backend startups Esplora and Electrum started peers and event processing before their first Lightning synchronization. With background sync disabled, startup could leave restored Lightning state stale indefinitely. Synchronize the Lightning wallet during initialization for both chain sources. Leave BDK synchronization to background or manual syncing and cover initial startup and restart against both backends. Co-Authored-By: HAL 9000 --- src/chain/mod.rs | 10 ++++-- src/lib.rs | 2 +- tests/integration_tests_rust.rs | 62 +++++++++++++++++++++++++++++++++ 3 files changed, 70 insertions(+), 4 deletions(-) diff --git a/src/chain/mod.rs b/src/chain/mod.rs index 7e1c908bed..0d2c981e9d 100644 --- a/src/chain/mod.rs +++ b/src/chain/mod.rs @@ -315,7 +315,7 @@ impl ChainSource { output_sweeper: Arc, ) -> Result<(), Error> { #[cfg(not(feature = "chain-bitcoind"))] - let _ = (stop_sync_receiver, onchain_wallet, channel_manager, chain_monitor, output_sweeper); + let _ = (stop_sync_receiver, onchain_wallet); match &self.kind { #[cfg(feature = "chain-bitcoind")] ChainSourceKind::Bitcoind(source) => { @@ -331,9 +331,13 @@ impl ChainSource { .await }, #[cfg(feature = "chain-esplora")] - ChainSourceKind::Esplora(_) => Ok(()), + ChainSourceKind::Esplora(source) => { + source.sync_lightning_wallet(channel_manager, chain_monitor, output_sweeper).await + }, #[cfg(feature = "chain-electrum")] - ChainSourceKind::Electrum(_) => Ok(()), + ChainSourceKind::Electrum(source) => { + source.sync_lightning_wallet(channel_manager, chain_monitor, output_sweeper).await + }, } } diff --git a/src/lib.rs b/src/lib.rs index 876257bfc6..b4e5e68c83 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -369,7 +369,7 @@ impl Node { ) })?; - // Reconcile block-based listeners before peers or event processing can act on stale + // Synchronize Lightning chain state before peers or event processing can act on stale // confirmations. Each restored ChannelMonitor may have been persisted at a different tip. self.runtime.block_on(self.chain_source.initialize_wallets( self.stop_sender.subscribe(), diff --git a/tests/integration_tests_rust.rs b/tests/integration_tests_rust.rs index ea21221f68..96bee16da8 100644 --- a/tests/integration_tests_rust.rs +++ b/tests/integration_tests_rust.rs @@ -1037,6 +1037,68 @@ async fn split_underpaid_bolt11_payment() { assert_eq!(node_b_payments.first().unwrap().amount_msat, Some(half_amount_msat)); } +#[cfg(feature = "chain-esplora")] +#[tokio::test(flavor = "multi_thread", worker_threads = 1)] +async fn startup_syncs_lightning_wallet_esplora() { + let (bitcoind, electrsd) = setup_bitcoind_and_electrsd(); + do_test_startup_syncs_lightning_wallet( + TestChainSource::Esplora(&electrsd), + &bitcoind, + &electrsd, + ) + .await; +} + +#[cfg(feature = "chain-electrum")] +#[tokio::test(flavor = "multi_thread", worker_threads = 1)] +async fn startup_syncs_lightning_wallet_electrum() { + let (bitcoind, electrsd) = setup_bitcoind_and_electrsd(); + do_test_startup_syncs_lightning_wallet( + TestChainSource::Electrum(&electrsd), + &bitcoind, + &electrsd, + ) + .await; +} + +#[cfg(any(feature = "chain-esplora", feature = "chain-electrum"))] +async fn do_test_startup_syncs_lightning_wallet( + chain_source: TestChainSource<'_>, bitcoind: &BitcoinD, electrsd: &ElectrsD, +) { + let config = random_config(); + setup_builder!(builder, config.node_config); + // Disable background synchronization so only startup can advance the Lightning tip. + configure_chain_source(&chain_source, &mut builder, &config); + let node = builder.build(config.node_entropy.into()).unwrap(); + assert!(node.status().latest_lightning_wallet_sync_timestamp.is_none()); + assert!(node.status().latest_onchain_wallet_sync_timestamp.is_none()); + + // Check both the first start and restarting the same node after the chain advances. + for _ in 0..2 { + let previous_tip = node.status().current_best_block; + generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 1).await; + let chain_info = bitcoind.client.get_blockchain_info().unwrap(); + let expected_hash = + bitcoind.client.get_block_hash(chain_info.blocks as u64).unwrap().block_hash().unwrap(); + assert_ne!(previous_tip.block_hash, expected_hash); + + node.start().unwrap(); + let status = node.status(); + assert_eq!( + status.current_best_block.block_hash, expected_hash, + "startup must synchronize the Lightning wallet before returning", + ); + assert_eq!(status.current_best_block.height as i64, chain_info.blocks); + assert!(status.latest_lightning_wallet_sync_timestamp.is_some()); + assert!(status.latest_fee_rate_cache_update_timestamp.is_some()); + assert!( + status.latest_onchain_wallet_sync_timestamp.is_none(), + "startup must leave the BDK wallet for background or manual synchronization", + ); + node.stop().unwrap(); + } +} + #[tokio::test(flavor = "multi_thread", worker_threads = 1)] async fn start_stop_reinit() { let (bitcoind, electrsd) = setup_bitcoind_and_electrsd(); From fc380fddf7123525807ce992273777bdad13a0fc Mon Sep 17 00:00:00 2001 From: Elias Rohrer Date: Tue, 13 Oct 2026 11:58:58 +0200 Subject: [PATCH 5/5] Refresh fee estimates during initialization Startup separately awaited fee estimates and backend capability checks before initializing wallets. These checks must complete before replay can construct claim transactions using the fee estimator. Move the checks into wallet initialization while preserving their ordering. Run fee refresh and required 0FC validation concurrently, then synchronize chain state only after both succeed. Co-Authored-By: HAL 9000 --- src/chain/mod.rs | 11 ++++++++++- src/lib.rs | 16 +--------------- 2 files changed, 11 insertions(+), 16 deletions(-) diff --git a/src/chain/mod.rs b/src/chain/mod.rs index 0d2c981e9d..96ade69dc5 100644 --- a/src/chain/mod.rs +++ b/src/chain/mod.rs @@ -312,8 +312,17 @@ impl ChainSource { pub(crate) async fn initialize_wallets( &self, stop_sync_receiver: tokio::sync::watch::Receiver<()>, onchain_wallet: Arc, channel_manager: Arc, chain_monitor: Arc, - output_sweeper: Arc, + output_sweeper: Arc, zero_fee_commitments_support_required: bool, ) -> Result<(), Error> { + // Replay can broadcast claims, so refresh fees and validate the backend before syncing. + // TODO: drop 0FC chain source validation when support is ubiquitous. + tokio::try_join!( + self.update_fee_rate_estimates(), + self.validate_zero_fee_commitments_support_if_required( + zero_fee_commitments_support_required + ) + )?; + #[cfg(not(feature = "chain-bitcoind"))] let _ = (stop_sync_receiver, onchain_wallet); match &self.kind { diff --git a/src/lib.rs b/src/lib.rs index b4e5e68c83..9d7d5b604c 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -354,21 +354,6 @@ impl Node { || monitor_owns_any_0fc_channels || self.config.anchor_channels_config.enable_zero_fee_commitments; - // Block to ensure we update our fee rate cache once on startup. - // Also take this opportunity to make sure our chain source supports 0FC channels - // if they are enabled. - // - // TODO: drop 0FC chain source validation when support is ubiquitous - let chain_source = Arc::clone(&self.chain_source); - self.runtime.block_on(async move { - tokio::try_join!( - chain_source.update_fee_rate_estimates(), - chain_source.validate_zero_fee_commitments_support_if_required( - zero_fee_commitments_support_required - ) - ) - })?; - // Synchronize Lightning chain state before peers or event processing can act on stale // confirmations. Each restored ChannelMonitor may have been persisted at a different tip. self.runtime.block_on(self.chain_source.initialize_wallets( @@ -377,6 +362,7 @@ impl Node { Arc::clone(&self.channel_manager), Arc::clone(&self.chain_monitor), Arc::clone(&self.output_sweeper), + zero_fee_commitments_support_required, ))?; // Spawn background task continuously syncing onchain, lightning, and fee rate cache.