diff --git a/src/chain/bitcoind.rs b/src/chain/bitcoind.rs index 615d1945a..f7589671b 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,8 +1318,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, + } }, } } @@ -1462,10 +1505,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 { @@ -1473,23 +1529,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); + } } } 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