From 66d48d23e1833c52fe9cbbd1dab0b73b9e0a47c3 Mon Sep 17 00:00:00 2001 From: Elias Rohrer Date: Wed, 5 Aug 2026 14:58:00 +0200 Subject: [PATCH 1/2] Follow reorgs with bitcoind REST Bitcoin Core's REST headers endpoint cannot serve headers from a stale branch. This prevented REST-backed nodes from finding a common ancestor and following the replacement chain after a reorg. In #1006 we enabled bitcoind REST chain source support in CI, which exposed this failure. Co-Authored-By: HAL 9000 --- src/chain/bitcoind.rs | 9 +++++-- tests/reorg_test.rs | 59 ++++++++++++++++++++++++++++++++++++++++++- 2 files changed, 65 insertions(+), 3 deletions(-) diff --git a/src/chain/bitcoind.rs b/src/chain/bitcoind.rs index 615d1945a..d471a3fb7 100644 --- a/src/chain/bitcoind.rs +++ b/src/chain/bitcoind.rs @@ -1280,8 +1280,13 @@ impl BlockSource for BitcoindClient { BitcoindClient::Rpc { rpc_client, .. } => { rpc_client.get_header(header_hash, height_hint).await }, - BitcoindClient::Rest { rest_client, .. } => { - rest_client.get_header(header_hash, height_hint).await + BitcoindClient::Rest { rest_client, rpc_client, .. } => { + match rest_client.get_header(header_hash, height_hint).await { + Err(e) if e.kind() == BlockSourceErrorKind::Persistent => { + rpc_client.get_header(header_hash, height_hint).await + }, + result => result, + } }, } } diff --git a/tests/reorg_test.rs b/tests/reorg_test.rs index 132d9de96..6e6d2d278 100644 --- a/tests/reorg_test.rs +++ b/tests/reorg_test.rs @@ -2,17 +2,74 @@ mod common; use std::collections::HashMap; use bitcoin::Amount; +use electrsd::corepc_node::mtype::ChainTipsStatus; use ldk_node::payment::{PaymentDirection, PaymentKind}; use ldk_node::{Event, LightningBalance, PendingSweepBalance}; use proptest::prelude::prop; use proptest::proptest; +use serde_json::json; use crate::common::{ expect_event, exponential_backoff_poll, generate_blocks_and_wait, invalidate_blocks, open_channel, premine_and_distribute_funds, random_chain_source, random_config, - setup_bitcoind_and_electrsd, setup_node, wait_for_outpoint_spend, wait_for_tx, + setup_bitcoind_and_electrsd, setup_node, wait_for_outpoint_spend, wait_for_tx, TestChainSource, }; +#[test] +fn bitcoind_rest_follows_valid_reorg() { + let rt = tokio::runtime::Builder::new_multi_thread().enable_all().build().unwrap(); + rt.block_on(async { + let (bitcoind, electrsd) = setup_bitcoind_and_electrsd(); + let node = setup_node(&TestChainSource::BitcoindRestSync(&bitcoind), random_config()); + let (bitcoind, electrs) = (&bitcoind.client, &electrsd.client); + + generate_blocks_and_wait(bitcoind, electrs, 3).await; + node.sync_wallets().unwrap(); + let original_tip = node.status().current_best_block; + let fork_block_hash = bitcoind + .get_block_hash((original_tip.height - 1) as u64) + .expect("failed to get fork block hash") + .block_hash() + .expect("fork block hash should be present"); + + invalidate_blocks(bitcoind, 2); + generate_blocks_and_wait(bitcoind, electrs, 3).await; + let replacement_tip_hash = + bitcoind.best_block_hash().expect("failed to get replacement tip"); + let replacement_tip_height = + bitcoind.get_blockchain_info().expect("failed to get replacement tip height").blocks + as u32; + + let _: serde_json::Value = bitcoind + .call("reconsiderblock", &[json!(fork_block_hash)]) + .expect("failed to reconsider original branch"); + let chain_tips = bitcoind + .get_chain_tips() + .expect("failed to get chain tips") + .into_model() + .expect("failed to parse chain tips") + .0; + assert!(chain_tips.iter().any(|tip| { + tip.hash == original_tip.block_hash && tip.status == ChainTipsStatus::ValidFork + })); + assert!(chain_tips.iter().any(|tip| { + tip.hash == replacement_tip_hash && tip.status == ChainTipsStatus::Active + })); + + node.sync_wallets() + .expect("REST-backed node did not follow Bitcoin Core's replacement chain"); + let synced_tip = node.status().current_best_block; + assert_eq!( + synced_tip.block_hash, replacement_tip_hash, + "REST-backed node did not follow Bitcoin Core's replacement chain" + ); + assert_eq!( + synced_tip.height, replacement_tip_height, + "REST-backed node did not follow Bitcoin Core's replacement chain" + ); + }) +} + async fn wait_for_pending_sweep_balance( node: &ldk_node::Node, mut matches_balance: F, ) -> PendingSweepBalance From 357f6e663b1c356cd0b51858c1f5e79e05f78ffa Mon Sep 17 00:00:00 2001 From: Elias Rohrer Date: Thu, 6 Aug 2026 13:31:00 +0200 Subject: [PATCH 2/2] Retain bitcoind SPV client Retain the cache populated during initial listener synchronization so subsequent polls can find headers for stale forks when REST no longer serves them. Use weak listener handles to avoid creating an ownership cycle through the chain source. Co-Authored-By: HAL 9000 --- src/chain/bitcoind.rs | 132 ++++++++++++++++++++++++++++++------------ 1 file changed, 95 insertions(+), 37 deletions(-) diff --git a/src/chain/bitcoind.rs b/src/chain/bitcoind.rs index d471a3fb7..1d39e9c58 100644 --- a/src/chain/bitcoind.rs +++ b/src/chain/bitcoind.rs @@ -49,8 +49,12 @@ use crate::{Error, PersistedNodeMetrics}; const CHAIN_POLLING_INTERVAL_SECS: u64 = 2; const CHAIN_POLLING_TIMEOUT_SECS: u64 = 10; +type BitcoindSpvClient = + SpvClient, BitcoindClient>, Arc>; + pub(super) struct BitcoindChainSource { api_client: Arc, + spv_client: tokio::sync::Mutex>, latest_chain_tip: RwLock>, wallet_polling_status: Mutex, fee_estimator: Arc, @@ -74,9 +78,11 @@ impl BitcoindChainSource { )); let latest_chain_tip = RwLock::new(None); + let spv_client = tokio::sync::Mutex::new(None); let wallet_polling_status = Mutex::new(WalletSyncStatus::Completed); Self { api_client, + spv_client, latest_chain_tip, wallet_polling_status, fee_estimator, @@ -103,10 +109,12 @@ impl BitcoindChainSource { )); let latest_chain_tip = RwLock::new(None); + let spv_client = tokio::sync::Mutex::new(None); let wallet_polling_status = Mutex::new(WalletSyncStatus::Completed); Self { api_client, + spv_client, latest_chain_tip, wallet_polling_status, fee_estimator, @@ -210,7 +218,16 @@ impl BitcoindChainSource { ) .await { - Ok((_header_cache, chain_tip)) => { + 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!( @@ -415,19 +432,24 @@ impl BitcoindChainSource { &self, onchain_wallet: Arc, channel_manager: Arc, chain_monitor: Arc, output_sweeper: Arc, ) -> Result<(), Error> { - let latest_chain_tip_opt = self.latest_chain_tip.read().expect("lock").clone(); - let chain_tip = - if let Some(tip) = latest_chain_tip_opt { tip } else { self.poll_chain_tip().await? }; - - let chain_poller = ChainPoller::new(Arc::clone(&self.api_client), self.config.network); - let chain_listener = ChainListener { - onchain_wallet: Arc::clone(&onchain_wallet), - channel_manager: Arc::clone(&channel_manager), - chain_monitor: Arc::clone(&chain_monitor), - output_sweeper, - }; - let mut spv_client = - SpvClient::new(chain_tip, chain_poller, HeaderCache::new(), &chain_listener); + let mut spv_client_lock = self.spv_client.lock().await; + if spv_client_lock.is_none() { + let latest_chain_tip_opt = self.latest_chain_tip.read().expect("lock").clone(); + let chain_tip = if let Some(tip) = latest_chain_tip_opt { + tip + } else { + self.poll_chain_tip().await? + }; + *spv_client_lock = Some(self.new_spv_client( + chain_tip, + HeaderCache::new(), + Arc::clone(&onchain_wallet), + Arc::clone(&channel_manager), + chain_monitor, + output_sweeper, + )); + } + let spv_client = spv_client_lock.as_mut().expect("initialized above"); let now = SystemTime::now(); match spv_client.poll_best_tip().await { @@ -442,6 +464,7 @@ impl BitcoindChainSource { return Err(Error::TxSyncFailed); }, } + drop(spv_client_lock); let cur_height = channel_manager.current_best_block().height; @@ -485,6 +508,21 @@ impl BitcoindChainSource { Ok(()) } + fn new_spv_client( + &self, chain_tip: ValidatedBlockHeader, header_cache: HeaderCache, + onchain_wallet: Arc, channel_manager: Arc, + chain_monitor: Arc, output_sweeper: Arc, + ) -> BitcoindSpvClient { + let chain_poller = ChainPoller::new(Arc::clone(&self.api_client), self.config.network); + let chain_listener = Arc::new(ChainListener { + onchain_wallet: Arc::downgrade(&onchain_wallet), + channel_manager: Arc::downgrade(&channel_manager), + chain_monitor: Arc::downgrade(&chain_monitor), + output_sweeper: Arc::downgrade(&output_sweeper), + }); + SpvClient::new(chain_tip, chain_poller, header_cache, chain_listener) + } + pub(super) async fn update_fee_rate_estimates(&self) -> Result<(), Error> { macro_rules! get_fee_rate_update { ($estimation_fut:expr) => {{ @@ -1280,13 +1318,8 @@ impl BlockSource for BitcoindClient { BitcoindClient::Rpc { rpc_client, .. } => { rpc_client.get_header(header_hash, height_hint).await }, - BitcoindClient::Rest { rest_client, rpc_client, .. } => { - match rest_client.get_header(header_hash, height_hint).await { - Err(e) if e.kind() == BlockSourceErrorKind::Persistent => { - rpc_client.get_header(header_hash, height_hint).await - }, - result => result, - } + BitcoindClient::Rest { rest_client, .. } => { + rest_client.get_header(header_hash, height_hint).await }, } } @@ -1467,10 +1500,23 @@ pub(crate) enum FeeRateEstimationMode { } pub(crate) struct ChainListener { - pub(crate) onchain_wallet: Arc, - pub(crate) channel_manager: Arc, - pub(crate) chain_monitor: Arc, - pub(crate) output_sweeper: Arc, + pub(crate) onchain_wallet: std::sync::Weak, + pub(crate) channel_manager: std::sync::Weak, + pub(crate) chain_monitor: std::sync::Weak, + pub(crate) output_sweeper: std::sync::Weak, +} + +impl ChainListener { + fn upgrade( + &self, + ) -> Option<(Arc, Arc, Arc, Arc)> { + Some(( + self.onchain_wallet.upgrade()?, + self.channel_manager.upgrade()?, + self.chain_monitor.upgrade()?, + self.output_sweeper.upgrade()?, + )) + } } impl Listen for ChainListener { @@ -1478,23 +1524,35 @@ impl Listen for ChainListener { &self, header: &bitcoin::block::Header, txdata: &lightning::chain::transaction::TransactionData, height: u32, ) { - self.onchain_wallet.filtered_block_connected(header, txdata, height); - self.channel_manager.filtered_block_connected(header, txdata, height); - self.chain_monitor.filtered_block_connected(header, txdata, height); - self.output_sweeper.filtered_block_connected(header, txdata, height); + if let Some((onchain_wallet, channel_manager, chain_monitor, output_sweeper)) = + self.upgrade() + { + onchain_wallet.filtered_block_connected(header, txdata, height); + channel_manager.filtered_block_connected(header, txdata, height); + chain_monitor.filtered_block_connected(header, txdata, height); + output_sweeper.filtered_block_connected(header, txdata, height); + } } fn block_connected(&self, block: &bitcoin::Block, height: u32) { - self.onchain_wallet.block_connected(block, height); - self.channel_manager.block_connected(block, height); - self.chain_monitor.block_connected(block, height); - self.output_sweeper.block_connected(block, height); + if let Some((onchain_wallet, channel_manager, chain_monitor, output_sweeper)) = + self.upgrade() + { + onchain_wallet.block_connected(block, height); + channel_manager.block_connected(block, height); + chain_monitor.block_connected(block, height); + output_sweeper.block_connected(block, height); + } } fn blocks_disconnected(&self, fork_point_block: lightning::chain::BlockLocator) { - self.onchain_wallet.blocks_disconnected(fork_point_block); - self.channel_manager.blocks_disconnected(fork_point_block); - self.chain_monitor.blocks_disconnected(fork_point_block); - self.output_sweeper.blocks_disconnected(fork_point_block); + if let Some((onchain_wallet, channel_manager, chain_monitor, output_sweeper)) = + self.upgrade() + { + onchain_wallet.blocks_disconnected(fork_point_block); + channel_manager.blocks_disconnected(fork_point_block); + chain_monitor.blocks_disconnected(fork_point_block); + output_sweeper.blocks_disconnected(fork_point_block); + } } }