diff --git a/src/chain/bitcoind.rs b/src/chain/bitcoind.rs index 9ed38c2128..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; @@ -52,6 +54,80 @@ const CHAIN_POLLING_TIMEOUT_SECS: u64 = 10; type BitcoindSpvClient = SpvClient, BitcoindClient>, Arc>; +#[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<()>, @@ -178,11 +254,12 @@ 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, - ) { + 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. let Some(initial_sync_guard) = @@ -190,7 +267,7 @@ impl BitcoindChainSource { .await else { log_trace!(self.logger, "Stopping initial chain sync."); - return; + return Err(Error::TxSyncFailed); }; log_info!( @@ -198,138 +275,83 @@ 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; + // 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); + } - 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; - } + 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 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 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)), + ]; - 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_chain_listeners( + self.api_client.as_ref(), + self.config.network, + &chain_monitor, + tx_broadcaster, + &self.fee_estimator, + &self.logger, + 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; - } - _ = 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; - } - _ = 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. 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, + ) { 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..6526c6137e --- /dev/null +++ b/src/chain/bitcoind/startup_tests.rs @@ -0,0 +1,310 @@ +// 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::collections::HashMap; +use std::sync::Arc; + +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 = + 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()); +} + +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 f01c1c8cb8..96ade69dc5 100644 --- a/src/chain/mod.rs +++ b/src/chain/mod.rs @@ -309,6 +309,47 @@ 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, 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 { + #[cfg(feature = "chain-bitcoind")] + ChainSourceKind::Bitcoind(source) => { + source + .initialize_wallets( + stop_sync_receiver, + onchain_wallet, + channel_manager, + chain_monitor, + output_sweeper, + &self.tx_broadcaster, + ) + .await + }, + #[cfg(feature = "chain-esplora")] + ChainSourceKind::Esplora(source) => { + source.sync_lightning_wallet(channel_manager, chain_monitor, output_sweeper).await + }, + #[cfg(feature = "chain-electrum")] + ChainSourceKind::Electrum(source) => { + source.sync_lightning_wallet(channel_manager, chain_monitor, output_sweeper).await + }, + } + } + 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..9d7d5b604c 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -354,20 +354,16 @@ 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( + self.stop_sender.subscribe(), + Arc::clone(&self.wallet), + 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. let stop_sync_receiver = 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();