diff --git a/Cargo.toml b/Cargo.toml index 7fc945833..520dcec4d 100755 --- a/Cargo.toml +++ b/Cargo.toml @@ -58,6 +58,7 @@ bdk_chain = { version = "0.23.3", default-features = false, features = ["std"] } bdk_esplora = { version = "0.22.2", default-features = false, features = ["async-https-rustls", "tokio"]} bdk_electrum = { version = "0.24.0", default-features = false, features = ["use-rustls-ring"]} bdk_wallet = { version = "3.1.0", default-features = false, features = ["std", "keys-bip39"]} +bip157 = { version = "0.6.1", default-features = false } bitreq = { version = "0.3", default-features = false, features = ["async-https", "json-using-serde"] } rustls = { version = "0.23", default-features = false } diff --git a/src/builder.rs b/src/builder.rs index a70b04b2a..616e6ff5c 100644 --- a/src/builder.rs +++ b/src/builder.rs @@ -45,7 +45,7 @@ use lightning::util::sweep::OutputSweeper; use lightning_dns_resolver::OMDomainResolver; use vss_client::headers::VssHeaderProvider; -use crate::chain::ChainSource; +use crate::chain::{CbfFeeSourceConfig, ChainSource}; use crate::config::{ default_user_config, may_announce_channel, AnnounceError, AsyncPaymentsRole, BitcoindRestClientConfig, Config, ElectrumSyncConfig, EsploraSyncConfig, HRNResolverConfig, @@ -112,6 +112,10 @@ enum ChainDataSourceConfig { rest_client_config: Option, wallet_rescan_from_height: Option, }, + Cbf { + peers: Vec, + fee_source_config: Option, + }, } #[derive(Debug, Clone)] @@ -396,6 +400,19 @@ impl NodeBuilder { self } + /// Configures the [`Node`] instance to source chain data via compact block filters + /// (BIP157/BIP158), connecting to the given peers (`ip:port`). + /// + /// `fee_source_config` optionally delegates fee estimation to an Esplora or Electrum server; + /// if `None`, fee rates are derived from recent blocks. + pub fn set_chain_source_cbf( + &mut self, peers: Vec, fee_source_config: Option, + ) -> &mut Self { + self.chain_data_source_config = + Some(ChainDataSourceConfig::Cbf { peers, fee_source_config }); + self + } + /// Configures the [`Node`] instance to connect to a Bitcoin Core node via RPC. /// /// This method establishes an RPC connection that enables all essential chain operations including @@ -1514,6 +1531,18 @@ fn build_with_store_internal( Arc::clone(&node_metrics), ) }, + Some(ChainDataSourceConfig::Cbf { peers, fee_source_config }) => ChainSource::new_cbf( + peers.clone(), + fee_source_config.clone(), + Arc::clone(&runtime), + Arc::clone(&fee_estimator), + Arc::clone(&tx_broadcaster), + Arc::clone(&kv_store), + Arc::clone(&config), + Arc::clone(&logger), + Arc::clone(&node_metrics), + ) + .map_err(|_| BuildError::ChainSourceSetupFailed)?, Some(ChainDataSourceConfig::Bitcoind { rpc_host, rpc_port, diff --git a/src/chain/bitcoind.rs b/src/chain/bitcoind.rs index f857ef533..75e986965 100644 --- a/src/chain/bitcoind.rs +++ b/src/chain/bitcoind.rs @@ -1458,6 +1458,7 @@ pub(crate) enum FeeRateEstimationMode { Conservative, } +#[derive(Clone)] pub(crate) struct ChainListener { pub(crate) onchain_wallet: Arc, pub(crate) channel_manager: Arc, @@ -1465,6 +1466,30 @@ pub(crate) struct ChainListener { pub(crate) output_sweeper: Arc, } +impl ChainListener { + pub(crate) fn get_best_block(&self) -> BlockLocator { + let candidates = [ + self.onchain_wallet.current_best_block(), + self.channel_manager.current_best_block(), + self.output_sweeper.current_best_block(), + ]; + let mut min = candidates.into_iter().min_by_key(|b| b.height).expect("non-empty"); + if let Some(worst_monitor) = self + .chain_monitor + .list_monitors() + .iter() + .flat_map(|id| self.chain_monitor.get_monitor(*id)) + .map(|m| m.current_best_block()) + .min_by_key(|b| b.height) + { + if worst_monitor.height < min.height { + min = worst_monitor; + } + } + min + } +} + impl Listen for ChainListener { fn filtered_block_connected( &self, header: &bitcoin::block::Header, diff --git a/src/chain/cbf.rs b/src/chain/cbf.rs new file mode 100644 index 000000000..7649370d7 --- /dev/null +++ b/src/chain/cbf.rs @@ -0,0 +1,1032 @@ +use std::collections::{BTreeMap, HashMap, HashSet}; +use std::net::SocketAddr; +use std::sync::{Arc, Mutex}; +use std::time::{Duration, SystemTime, UNIX_EPOCH}; + +use bip157::chain::ChainState; +use bip157::{ + chain::BlockHeaderChanges, Builder as KyotoBuilder, Client, Event, HashCheckpoint, Header, + IndexedBlock, Info, Node as KyotoNode, Package, Requester, TrustedPeer, Warning, +}; +use bitcoin::{BlockHash, FeeRate, Network, Script, ScriptBuf, Transaction, Txid}; +use electrum_client::{Client as ElectrumClient, ConfigBuilder as ElectrumConfigBuilder}; +use lightning::chain::{BlockLocator, Listen, WatchedOutput}; + +use tokio::sync::{mpsc, watch}; + +use crate::chain::bitcoind::ChainListener; +use crate::chain::electrum::get_electrum_fee_rate_cache_update; +use crate::chain::CbfFeeSourceConfig; +use crate::config::{Config, DEFAULT_FEE_RATE_CACHE_UPDATE_INTERVAL_SECS}; +use crate::error::Error; +use crate::fee_estimator::{ + apply_post_estimation_adjustments, get_all_conf_targets, get_fallback_rate_for_target, + get_num_block_defaults_for_target, ConfirmationTarget, OnchainFeeEstimator, +}; +use crate::io::utils::update_and_persist_node_metrics; +use crate::logger::{log_debug, log_error, log_info, log_trace, LdkLogger, Logger}; +use crate::runtime::Runtime; +use crate::types::DynStore; +use crate::util::{cbf_percentile_for_target, coinbase_fee_rate, percentile_of_sorted}; +use crate::wallet::Wallet; +use crate::PersistedNodeMetrics; + +/// Walk back this many blocks from the wallet's persisted tip when deriving +/// the kyoto resume checkpoint, so a recent reorg cannot strand the node +/// above the new best chain. +const REORG_SAFETY_BLOCKS: u32 = 7; +const BLOCK_FEE_CACHE_CAPACITY: usize = REORG_SAFETY_BLOCKS as usize * 2; + +/// Peer response timeout passed to kyoto's `Builder::response_timeout`. +const DEFAULT_RESPONSE_TIMEOUT_SECS: u64 = 30; + +/// Number of peers that must agree on filter headers before they're accepted. +const DEFAULT_REQUIRED_PEERS: u8 = 1; + +/// Maximum consecutive `node.run()` failures before the restart loop gives up. +const MAX_RESTART_RETRIES: u32 = 5; + +/// Initial backoff delay between restart attempts; doubles each failure. +const INITIAL_BACKOFF_MS: u64 = 500; + +/// Retry matched block downloads before surfacing a CBF sync failure. +const CBF_BLOCK_FETCH_RETRIES: u8 = 3; + +const ESPLORA_TIMEOUT: u64 = 2; + +/// Retries and per-request timeout for the fresh Electrum connection opened each fee cycle. +const ELECTRUM_FEE_NUM_RETRIES: u8 = 3; +const ELECTRUM_FEE_TIMEOUT_SECS: u64 = 10; + +/// Runtime status of the underlying kyoto node. +enum CbfRuntimeStatus { + Started { requester: Requester }, + Stopped, +} + +#[derive(Clone, Copy)] +enum CbfSyncState { + Active { + /// Highest tip whose preceding chain updates have been applied to all listeners. + applied_tip: Option, + /// Whether kyoto has reported catching up to the network tip (via `FiltersSynced`) and + /// the resulting blocks have been applied. `wait_until_synced` blocks until this is set. + /// + /// This must not be derived from a locally-sampled chain tip: kyoto does not persist, so a + /// freshly (re)started node's local header chain sits at genesis until it syncs from peers. + /// Comparing against that would make `wait_until_synced` return before any sync happens. + synced_to_tip: bool, + }, + Failed(Error), +} + +/// Marks that we are applying a block past the last `FiltersSynced` tip, so a `sync_wallets` call +/// issued after new blocks are mined waits for the next `FiltersSynced` rather than returning on a +/// stale `synced_to_tip`. Only flips (and notifies waiters) when currently set. +/// +/// Called both when a new block's filter is received (before it is fetched and applied) and after +/// it is applied, so `synced_to_tip` reflects "behind by an unapplied block" as soon as we learn +/// that block exists, not only once we've finished catching up to it. +fn mark_syncing(sync_state_tx: &watch::Sender) { + // Copy the current state out and drop the `watch` read guard before calling `send_replace`: + // `borrow()` holds a read lock for the lifetime of its temporary, and `send_replace` takes + // the write lock, so holding the borrow across it deadlocks. `CbfSyncState` is `Copy`, so the + // deref copies and the guard is released at the end of this statement. + let current = *sync_state_tx.borrow(); + if let CbfSyncState::Active { applied_tip, synced_to_tip: true } = current { + sync_state_tx.send_replace(CbfSyncState::Active { applied_tip, synced_to_tip: false }); + } +} + +/// Struct for holding cbf chain source +pub struct CbfChainSource { + /// Trusted peer addresses for kyoto's `Builder::add_peers`. + trusted_peers: Vec, + /// Scripts tracked by LDK, onchain wallet's scripts are pulled from the onchain wallet + registered_scripts: Arc>>, + fee_source: FeeSource, + /// Tracks whether the kyoto node is running and holds the live requester. + cbf_runtime_status: Arc>, + /// Highest CBF sync tip whose preceding chain updates have been applied to all listeners. + sync_state_tx: watch::Sender, + /// Handle used to spawn the background tasks and offload blocking work. + runtime: Arc, + /// Node configuration (network, storage path). + config: Arc, + fee_estimator: Arc, + kv_store: Arc, + node_metrics: Arc, + logger: Arc, +} + +#[derive(Debug)] +enum ChainOp { + ConnectFull { block: IndexedBlock }, + ConnectFiltered { header: Header, height: u32 }, + Disconnect { fork_point: BlockLocator }, + Synced { tip_height: u32 }, + Failed { error: Error }, +} + +struct BlockApplicator { + chain_listener: ChainListener, + ops_rx: mpsc::UnboundedReceiver, + next_height: u32, + sync_state_tx: watch::Sender, + /// Present only for the native CBF fee source: lets us cache the fee rate of blocks we download + /// here, so the fee estimator doesn't have to re-download them. + block_fee_cache: Option, + kv_store: Arc, + node_metrics: Arc, + logger: Arc, +} + +impl BlockApplicator { + async fn run(mut self) { + while let Some(op) = self.ops_rx.recv().await { + match op { + ChainOp::ConnectFull { block: ib } => { + if ib.height != self.next_height { + log_debug!( + self.logger, + "CBF skipping out-of-sequence block at height {} (expected {})", + ib.height, + self.next_height + ); + continue; + } + self.chain_listener.block_connected(&ib.block, ib.height); + self.next_height += 1; + mark_syncing(&self.sync_state_tx); + if let Some(cache) = &self.block_fee_cache { + let fee_rate = coinbase_fee_rate(&ib.block, ib.height); + cache + .lock() + .expect("lock") + .insert(ib.height, (ib.block.block_hash(), fee_rate)); + } + }, + ChainOp::ConnectFiltered { header, height } => { + if height != self.next_height { + log_debug!( + self.logger, + "CBF skipping out-of-sequence block at height {} (expected {})", + height, + self.next_height + ); + continue; + } + self.chain_listener.filtered_block_connected(&header, &[], height); + self.next_height += 1; + mark_syncing(&self.sync_state_tx); + }, + ChainOp::Disconnect { fork_point } => { + self.chain_listener.blocks_disconnected(fork_point); + self.next_height = fork_point.height + 1; + self.sync_state_tx.send_replace(CbfSyncState::Active { + applied_tip: Some(fork_point.height), + synced_to_tip: false, + }); + }, + ChainOp::Synced { tip_height } => { + log_info!(self.logger, "CBF caught up to tip {}", tip_height); + if self.next_height > tip_height { + self.publish_synced_tip(tip_height).await; + } else { + log_debug!( + self.logger, + "CBF waiting to apply blocks through tip {} (next height {})", + tip_height, + self.next_height + ); + } + log_info!(self.logger, "we set new tip and published at {}", tip_height); + }, + ChainOp::Failed { error } => { + log_info!(self.logger, "we received error chain op {}", error); + self.sync_state_tx.send_replace(CbfSyncState::Failed(error)); + }, + } + } + } + + async fn publish_synced_tip(&self, tip_height: u32) { + let already_published = { + let sync_state = *self.sync_state_tx.borrow(); + match sync_state { + CbfSyncState::Active { applied_tip, .. } => applied_tip, + CbfSyncState::Failed(_) => None, + } + }; + if already_published.map_or(false, |published_height| published_height >= tip_height) { + // Even if the applied tip is unchanged, we have now confirmed we are caught up to the + // network tip, so ensure the synced flag is set for any `wait_until_synced` waiter. + self.sync_state_tx.send_replace(CbfSyncState::Active { + applied_tip: already_published, + synced_to_tip: true, + }); + return; + } + self.sync_state_tx.send_replace(CbfSyncState::Active { + applied_tip: Some(tip_height), + synced_to_tip: true, + }); + let unix_time_secs_opt = + SystemTime::now().duration_since(UNIX_EPOCH).ok().map(|d| d.as_secs()); + if let Err(e) = 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 + { + log_error!(self.logger, "Failed to persist CBF sync metrics: {:?}", e); + } + } +} + +/// Number of most recent blocks whose coinbase-derived fee rates feed the native CBF estimator. +const FEE_WINDOW_BLOCKS: u32 = BLOCK_FEE_CACHE_CAPACITY as u32; + +/// Lower bound for native CBF fee estimates (1 sat/vB), matching the floor used by the Esplora and +/// Electrum fee sources. Coinbase-derived rates are frequently zero on regtest/signet. +const CBF_MIN_FEERATE_SAT_PER_KWU: u64 = 250; + +/// Per-attempt timeout when downloading a block from a peer — used both for matched blocks we apply +/// to the listeners and for the coinbase-fee-rate samples. Kyoto queues the request and awaits a +/// peer response with no timeout of its own, so a slow or unresponsive peer would otherwise park the +/// fetch forever. Kept short so a single request is bounded and can be retried (or, for fees, only +/// delays one sample) rather than stalling. +const CBF_BLOCK_FETCH_TIMEOUT_SECS: u64 = 10; + +/// Recent per-block coinbase-derived fee rates, keyed by height so we can window on the tip, evict +/// stale entries, and detect reorged-out blocks (a height whose cached hash no longer matches the +/// canonical chain). Shared via `Arc` between the fee estimator and the [`BlockApplicator`]. +type BlockFeeCache = Arc>>; + +enum FeeSource { + /// Derive fee rates from the coinbase reward of recent blocks. Downloads full blocks in order to calculate fee estimation. + /// + /// The [`BlockApplicator`] also opportunistically inserts the fee rate of any block it already + /// downloads on a filter match, saving a re-download in the reconciliation loop. + Cbf { block_fee_cache: BlockFeeCache }, + /// Delegate fee estimation to an Esplora HTTP server. + Esplora { client: esplora_client::AsyncClient }, + /// Delegate fee estimation to an Electrum server. + /// + /// A fresh connection is opened for each estimation cycle. + Electrum { server_url: String }, +} + +impl CbfChainSource { + pub(crate) fn new( + peers: Vec, fee_source_config: Option, runtime: Arc, + fee_estimator: Arc, kv_store: Arc, config: Arc, + logger: Arc, node_metrics: Arc, + ) -> Result { + let trusted_peers: Vec = peers + .iter() + .filter_map(|peer_str| { + peer_str.parse::().ok().map(TrustedPeer::from_socket_addr) + }) + .collect(); + + let fee_source = match fee_source_config { + Some(CbfFeeSourceConfig::Esplora(server_url)) => { + let mut esplora_builder = esplora_client::Builder::new(&server_url); + esplora_builder = esplora_builder.timeout(ESPLORA_TIMEOUT); + let client = esplora_builder.build_async().map_err(|e| { + log_error!(logger, "Failed to build esplora client: {}", e); + Error::ConnectionFailed + })?; + FeeSource::Esplora { client } + }, + Some(CbfFeeSourceConfig::Electrum(server_url)) => FeeSource::Electrum { server_url }, + None => FeeSource::Cbf { block_fee_cache: Arc::new(Mutex::new(BTreeMap::new())) }, + }; + let registered_scripts = Arc::new(Mutex::new(HashSet::new())); + let cbf_runtime_status = Arc::new(Mutex::new(CbfRuntimeStatus::Stopped)); + let (sync_state_tx, _) = + watch::channel(CbfSyncState::Active { applied_tip: None, synced_to_tip: false }); + Ok(Self { + trusted_peers, + fee_source, + registered_scripts, + cbf_runtime_status, + sync_state_tx, + runtime, + config, + fee_estimator, + kv_store, + node_metrics, + logger, + }) + } + + fn build_kyoto( + trusted_peers: &[TrustedPeer], config: &Config, logger: &Logger, + chain_listener: &ChainListener, + ) -> (KyotoNode, Client) { + let mut kyoto_builder = KyotoBuilder::new(config.network); + + let data_dir = std::path::PathBuf::from(&config.storage_dir_path).join("bip157_data"); + kyoto_builder = kyoto_builder.data_dir(data_dir); + + if !trusted_peers.is_empty() { + kyoto_builder = kyoto_builder.add_peers(trusted_peers.to_vec()); + } + + kyoto_builder = kyoto_builder.required_peers(DEFAULT_REQUIRED_PEERS); + kyoto_builder = kyoto_builder.fetch_witness_data(); + kyoto_builder = + kyoto_builder.response_timeout(Duration::from_secs(DEFAULT_RESPONSE_TIMEOUT_SECS)); + + if let Some(header_cp) = resume_checkpoint(logger, chain_listener) { + log_debug!( + logger, + "CBF builder: resuming from checkpoint height={}, hash={}", + header_cp.height, + header_cp.hash, + ); + kyoto_builder = kyoto_builder.chain_state(ChainState::Checkpoint(header_cp)); + } + + kyoto_builder.build() + } + + pub(crate) fn start(&self, chain_listener: ChainListener) { + let (node, client) = + Self::build_kyoto(&self.trusted_peers, &self.config, &self.logger, &chain_listener); + let Client { requester, info_rx, warn_rx, event_rx } = client; + + { + let mut status = self.cbf_runtime_status.lock().expect("lock"); + if matches!(*status, CbfRuntimeStatus::Started { .. }) { + debug_assert!(false, "start() called while CBF chain source is already running"); + return; + } + *status = CbfRuntimeStatus::Started { requester }; + } + + let (ops_tx, ops_rx) = mpsc::unbounded_channel(); + let block_fee_cache = match &self.fee_source { + FeeSource::Cbf { block_fee_cache } => Some(Arc::clone(block_fee_cache)), + _ => None, + }; + let best_block_height = chain_listener.get_best_block().height; + self.sync_state_tx.send_replace(CbfSyncState::Active { + applied_tip: Some(best_block_height), + synced_to_tip: false, + }); + let block_applicator = BlockApplicator { + next_height: best_block_height + 1, + sync_state_tx: self.sync_state_tx.clone(), + chain_listener: chain_listener.clone(), + ops_rx, + block_fee_cache, + kv_store: Arc::clone(&self.kv_store), + node_metrics: Arc::clone(&self.node_metrics), + logger: Arc::clone(&self.logger), + }; + self.runtime.spawn_background_task(block_applicator.run()); + + log_info!(self.logger, "CBF chain source started."); + + let restart_status = Arc::clone(&self.cbf_runtime_status); + let restart_logger = Arc::clone(&self.logger); + let restart_peers = self.trusted_peers.clone(); + let restart_config = Arc::clone(&self.config); + let restart_listener = chain_listener; + let restart_registered_scripts = Arc::clone(&self.registered_scripts); + let restart_cbf_runtime_status = Arc::clone(&self.cbf_runtime_status); + let restart_sync_state_tx = self.sync_state_tx.clone(); + + self.runtime.spawn_background_task(async move { + let mut current_node = node; + let mut current_info_rx = info_rx; + let mut current_warn_rx = warn_rx; + let mut current_event_rx = event_rx; + let mut retries = 0u32; + let mut backoff_ms = INITIAL_BACKOFF_MS; + + loop { + let info_handle = tokio::spawn(Self::process_info_messages( + current_info_rx, + Arc::clone(&restart_logger), + )); + let warn_handle = tokio::spawn(Self::process_warn_messages( + current_warn_rx, + Arc::clone(&restart_logger), + )); + + let event_handle = tokio::spawn(Self::process_kyoto_events( + Arc::clone(&restart_logger), + current_event_rx, + Arc::clone(&restart_registered_scripts), + Arc::clone(&restart_cbf_runtime_status), + ops_tx.clone(), + Arc::clone(&restart_listener.onchain_wallet), + restart_sync_state_tx.clone(), + )); + + match current_node.run().await { + Ok(()) => { + log_info!(restart_logger, "CBF node shut down cleanly."); + *restart_status.lock().expect("lock") = CbfRuntimeStatus::Stopped; + restart_sync_state_tx.send_replace(CbfSyncState::Failed(Error::NotRunning)); + break; + }, + Err(e) => { + retries += 1; + if retries > MAX_RESTART_RETRIES { + log_error!( + restart_logger, + "CBF node failed {} times, giving up: {:?}", + retries, + e, + ); + *restart_status.lock().expect("lock") = CbfRuntimeStatus::Stopped; + restart_sync_state_tx + .send_replace(CbfSyncState::Failed(Error::TxSyncFailed)); + break; + } + log_error!( + restart_logger, + "CBF node exited with error (attempt {}/{}): {:?}. Restarting in {}ms.", + retries, + MAX_RESTART_RETRIES, + e, + backoff_ms, + ); + + // Abort the old log consumers before rebuilding. + info_handle.abort(); + warn_handle.abort(); + event_handle.abort(); + + tokio::time::sleep(Duration::from_millis(backoff_ms)).await; + backoff_ms = backoff_ms.saturating_mul(2); + let (new_node, new_client) = Self::build_kyoto( + &restart_peers, + &restart_config, + &restart_logger, + &restart_listener, + ); + let Client { + requester: new_requester, + info_rx: new_info_rx, + warn_rx: new_warn_rx, + event_rx: new_event_rx, + } = new_client; + + { + let mut status = restart_status.lock().expect("lock"); + if matches!(*status, CbfRuntimeStatus::Stopped) { + let _ = new_requester.shutdown(); + restart_sync_state_tx + .send_replace(CbfSyncState::Failed(Error::NotRunning)); + log_info!( + restart_logger, + "CBF restart aborted: stop() called during backoff." + ); + break; + } + *status = CbfRuntimeStatus::Started { requester: new_requester }; + restart_sync_state_tx.send_replace(CbfSyncState::Active { + applied_tip: Some(restart_listener.get_best_block().height), + synced_to_tip: false, + }); + } + + current_node = new_node; + current_info_rx = new_info_rx; + current_warn_rx = new_warn_rx; + current_event_rx = new_event_rx; + }, + } + } + }); + } + + pub(crate) fn stop(&self) { + let requester = { + let mut status = self.cbf_runtime_status.lock().expect("lock"); + match &*status { + CbfRuntimeStatus::Started { requester } => { + let requester = requester.clone(); + *status = CbfRuntimeStatus::Stopped; + Some(requester) + }, + CbfRuntimeStatus::Stopped => None, + } + }; + + if let Some(requester) = requester { + if let Err(e) = requester.shutdown() { + log_error!(self.logger, "Failed to shut down CBF node: {:?}", e); + } + } + self.sync_state_tx.send_replace(CbfSyncState::Failed(Error::NotRunning)); + } + + pub(crate) async fn wait_until_synced(&self) -> Result<(), Error> { + if matches!(&*self.cbf_runtime_status.lock().expect("lock"), CbfRuntimeStatus::Stopped) { + return Err(Error::NotRunning); + } + let mut sync_state_rx = self.sync_state_tx.subscribe(); + + // Wait for kyoto to report catching up to the network tip (a `FiltersSynced`-driven + // `synced_to_tip`) and for the resulting blocks to be applied. We must not target a + // locally-sampled `chain_tip()`: kyoto does not persist, so a freshly (re)started node's + // local header chain sits at genesis until it syncs from peers, which would let this return + // before any sync happens. + loop { + match *sync_state_rx.borrow() { + CbfSyncState::Active { synced_to_tip, .. } => { + if synced_to_tip { + return Ok(()); + } + }, + CbfSyncState::Failed(error) => return Err(error), + } + + if let Err(e) = sync_state_rx.changed().await { + debug_assert!(false, "Failed to receive CBF sync result: {:?}", e); + log_error!(self.logger, "Failed to receive CBF sync result: {:?}", e); + return Err(Error::TxSyncFailed); + } + } + } + + async fn process_info_messages(mut info_rx: mpsc::Receiver, logger: Arc) { + while let Some(info) = info_rx.recv().await { + log_debug!(logger, "CBF node info: {}", info); + } + } + + async fn process_warn_messages( + mut warn_rx: mpsc::UnboundedReceiver, logger: Arc, + ) { + while let Some(warning) = warn_rx.recv().await { + log_debug!(logger, "CBF node warning: {}", warning); + } + } + + async fn process_kyoto_events( + logger: Arc, mut event_rx: mpsc::UnboundedReceiver, + registered_scripts: Arc>>, + cbf_runtime_status: Arc>, ops_tx: mpsc::UnboundedSender, + onchain_wallet: Arc, sync_state_tx: watch::Sender, + ) { + while let Some(event) = event_rx.recv().await { + match event { + Event::IndexedFilter(indexed_filter) => { + // A new block's filter arrived, so we're behind by at least this block until it + // is fetched (if matched) and applied. Flip this before the fetch, not after, + // so a `sync_wallets` call issued in between doesn't return on a stale + // `synced_to_tip` that predates this block. + mark_syncing(&sync_state_tx); + + let requester = match &*cbf_runtime_status.lock().expect("lock") { + CbfRuntimeStatus::Started { requester } => requester.clone(), + CbfRuntimeStatus::Stopped => { + let _ = ops_tx.send(ChainOp::Failed { error: Error::NotRunning }); + return; + }, + }; + //registered_scripts contains only LDK scripts, not onchain wallet's scripts, + //as don't want to track them twice: once in bdk, once in CbfChainSource, thus + //each time we receive an IndexedFilter event, we ask bdk to give us all + //revealed scripts. We create all_scripts starting from onchain wallet's + //scripts and extend them with LDK's ones + let mut all_scripts = onchain_wallet.list_watched_scripts(); + all_scripts.extend(registered_scripts.lock().expect("lock").iter().cloned()); + + let block_hash = indexed_filter.block_hash(); + let matched = indexed_filter.contains_any(all_scripts.iter()); + + let chop: ChainOp = if matched { + let mut attempt = 0; + let block = loop { + attempt += 1; + let handle = match requester.request_block(block_hash) { + Ok(handle) => handle, + Err(_) => { + log_error!( + logger, + "Failed to obtain receiver for matched CBF block {}; node is stopped", + block_hash + ); + let _ = + ops_tx.send(ChainOp::Failed { error: Error::NotRunning }); + return; + }, + }; + + // Bound the download so an unresponsive peer can't park the fetch forever, + // then flatten the three error layers (timeout / receiver dropped / fetch + // error) into a single reason so the retry-or-fail decision is written once. + let fetched = tokio::time::timeout( + Duration::from_secs(CBF_BLOCK_FETCH_TIMEOUT_SECS), + handle, + ) + .await + .map_err(|_| { + format!("timed out after {}s", CBF_BLOCK_FETCH_TIMEOUT_SECS) + }) + .and_then(|recv| recv.map_err(|_| "receiver was dropped".to_string())) + .and_then(|fetch| fetch.map_err(|e| format!("failed: {:?}", e))); + + match fetched { + Ok(block) => break block, + Err(reason) if attempt < CBF_BLOCK_FETCH_RETRIES => { + log_debug!( + logger, + "CBF block fetch for {} {} on attempt {}; retrying", + block_hash, + reason, + attempt + ); + }, + Err(reason) => { + log_error!( + logger, + "CBF block fetch for {} {} after {} attempts; giving up", + block_hash, + reason, + CBF_BLOCK_FETCH_RETRIES + ); + let _ = + ops_tx.send(ChainOp::Failed { error: Error::TxSyncFailed }); + return; + }, + } + }; + ChainOp::ConnectFull { block } + } else { + let height = indexed_filter.height(); + //TODO we need to recheck that a particular height has not been + //reorganized, and we retrieve indeed the same block header that we + //received `IndexedFilter` event of. + match requester.get_header(height).await { + Ok(Some(indexed_header)) => { + if indexed_header.block_hash() != block_hash { + log_debug!( + logger, + "Filter for {} reorged; skipping", + block_hash + ); + continue; + } + ChainOp::ConnectFiltered { + header: indexed_header.header, + height: indexed_header.height, + } + }, + Ok(None) => { + log_error!(logger, "No header at height {}", height,); + let _ = ops_tx.send(ChainOp::Failed { error: Error::TxSyncFailed }); + break; + }, + Err(e) => { + log_error!( + logger, + "Failed to fetch header at height {}: {:?}", + height, + e, + ); + let _ = ops_tx.send(ChainOp::Failed { error: Error::TxSyncFailed }); + break; + }, + } + }; + if let Err(e) = ops_tx.send(chop) { + log_debug!(logger, "ops_rx gone: {}", e); + } + }, + Event::FiltersSynced(sync_update) => { + //Because application of blocks is async, the fact that kyoto synced up to the + //tip does NOT mean that we caught everything up, that's why we send a ChainOp, + //only processing of which means we processed all blocks up to the tip. + log_info!(logger, "Kyoto synced up to the tip {}", sync_update.tip().height); + let _ = ops_tx.send(ChainOp::Synced { tip_height: sync_update.tip().height }); + }, + Event::ChainUpdate(BlockHeaderChanges::Connected(indexed_header)) => { + log_debug!( + logger, + "Kyoto connected header at height {}", + indexed_header.height + ); + }, + Event::ChainUpdate(BlockHeaderChanges::Reorganized { + reorganized, + accepted: _, + }) => { + // Rewind to the fork point; kyoto will re-deliver the new chain's filters. + if let Some(lowest) = reorganized.first() { + let fork_point = BlockLocator::new( + lowest.prev_blockhash(), + lowest.height.saturating_sub(1), + ); + let _ = ops_tx.send(ChainOp::Disconnect { fork_point }); + } + }, + Event::ChainUpdate(BlockHeaderChanges::ForkAdded(fork)) => { + log_debug!(logger, "Kyoto added fork header at height {}", fork.height); + }, + } + } + } + + pub(crate) fn register_tx(&self, _txid: &Txid, script_pubkey: &Script) { + self.registered_scripts.lock().expect("lock").insert(script_pubkey.into()); + } + + pub(crate) fn register_output(&self, output: WatchedOutput) { + self.registered_scripts.lock().expect("lock").insert(output.script_pubkey); + } + + pub(crate) async fn continuously_update_fee_rate_estimates( + &self, mut stop_sync_receiver: watch::Receiver<()>, + ) { + let mut fee_rate_update_interval = + tokio::time::interval(Duration::from_secs(DEFAULT_FEE_RATE_CACHE_UPDATE_INTERVAL_SECS)); + // We primed the cache once on startup, so skip the immediate first tick. + fee_rate_update_interval.reset(); + fee_rate_update_interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); + + loop { + tokio::select! { + _ = stop_sync_receiver.changed() => { + log_trace!(self.logger, "Stopping CBF fee-rate update loop."); + return; + } + _ = fee_rate_update_interval.tick() => { + if let Err(e) = self.update_fee_rate_estimates().await { + log_error!(self.logger, "Failed to update fee rate estimates: {:?}", e); + } + } + } + } + } + + pub(crate) async fn update_fee_rate_estimates(&self) -> Result<(), Error> { + let new_fee_rate_cache = match &self.fee_source { + FeeSource::Esplora { client } => { + let estimates = client.get_fee_estimates().await.map_err(|e| { + log_error!(self.logger, "Failed to retrieve fee rate estimates: {}", e); + Error::FeerateEstimationUpdateFailed + })?; + + if estimates.is_empty() && self.config.network == Network::Bitcoin { + log_error!( + self.logger, + "Failed to retrieve fee rate: empty fee estimates are disallowed on Mainnet." + ); + return Err(Error::FeerateEstimationUpdateFailed); + } + + let mut new_fee_rate_cache = HashMap::with_capacity(10); + for target in get_all_conf_targets() { + let num_blocks = get_num_block_defaults_for_target(target); + // Fall back to 1 sat/vb if we fail or it yields less than that, mostly to keep + // going on signet/regtest where estimates may be missing or bogus. + let converted_estimate_sat_vb = + esplora_client::convert_fee_rate(num_blocks, estimates.clone()) + .map_or(1.0, |converted| converted.max(1.0)); + let fee_rate = + FeeRate::from_sat_per_kwu((converted_estimate_sat_vb * 250.0) as u64); + let adjusted_fee_rate = apply_post_estimation_adjustments(target, fee_rate); + new_fee_rate_cache.insert(target, adjusted_fee_rate); + } + new_fee_rate_cache + }, + FeeSource::Electrum { server_url } => { + let electrum_config = ElectrumConfigBuilder::new() + .retry(ELECTRUM_FEE_NUM_RETRIES) + .timeout(Some(Duration::from_secs(ELECTRUM_FEE_TIMEOUT_SECS))) + .build(); + + let server_url = server_url.clone(); + let electrum_client = self + .runtime + .spawn_blocking(move || { + ElectrumClient::from_config(&server_url, electrum_config) + }) + .await + .map_err(|e| { + log_error!(self.logger, "Fee rate estimation task panicked: {}", e); + Error::FeerateEstimationUpdateFailed + })? + .map_err(|e| { + log_error!(self.logger, "Failed to connect to electrum server: {}", e); + Error::ConnectionFailed + })?; + + get_electrum_fee_rate_cache_update( + Arc::clone(&self.runtime), + Arc::new(electrum_client), + self.config.network, + ELECTRUM_FEE_TIMEOUT_SECS, + Arc::clone(&self.logger), + ) + .await? + }, + FeeSource::Cbf { block_fee_cache } => { + let requester = match &*self.cbf_runtime_status.lock().expect("lock") { + CbfRuntimeStatus::Started { requester } => requester.clone(), + CbfRuntimeStatus::Stopped => return Err(Error::FeerateEstimationUpdateFailed), + }; + let mut samples_sat_per_kwu: Vec = self + .refresh_block_fee_window(&requester, block_fee_cache) + .await + .iter() + .map(|rate| rate.to_sat_per_kwu()) + .collect(); + samples_sat_per_kwu.sort_unstable(); + + let mut new_fee_rate_cache = HashMap::with_capacity(10); + for target in get_all_conf_targets() { + let fee_rate = if samples_sat_per_kwu.is_empty() { + FeeRate::from_sat_per_kwu(get_fallback_rate_for_target(target) as u64) + } else { + let percentile = cbf_percentile_for_target(target); + let sat_per_kwu = percentile_of_sorted(&samples_sat_per_kwu, percentile) + .max(CBF_MIN_FEERATE_SAT_PER_KWU); + FeeRate::from_sat_per_kwu(sat_per_kwu) + }; + let adjusted_fee_rate = apply_post_estimation_adjustments(target, fee_rate); + new_fee_rate_cache.insert(target, adjusted_fee_rate); + } + new_fee_rate_cache + }, + }; + + self.commit_fee_rate_cache(new_fee_rate_cache).await + } + + /// Writes a freshly computed per-target fee-rate map into the estimator cache and records the + /// update timestamp in the node metrics. + async fn commit_fee_rate_cache( + &self, new_fee_rate_cache: HashMap, + ) -> Result<(), Error> { + self.fee_estimator.set_fee_rate_cache(new_fee_rate_cache); + 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_fee_rate_cache_update_timestamp = unix_time_secs_opt + }) + .await?; + Ok(()) + } + + pub(crate) async fn process_broadcast_package(&self, package: Vec) { + let requester = match &*self.cbf_runtime_status.lock().expect("lock") { + CbfRuntimeStatus::Started { requester } => requester.clone(), + CbfRuntimeStatus::Stopped => { + debug_assert!(false, "We should have started the chain source before broadcasting"); + return; + }, + }; + + match Package::from_vec(package.clone()) { + Ok(package) => { + if let Err(e) = requester.submit_package(package).await { + log_error!(self.logger, "Failed to broadcast transaction package: {:?}", e); + } + }, + Err(_) => { + for tx in package { + let txid = tx.compute_txid(); + if let Err(e) = requester.submit_package(tx).await { + log_error!( + self.logger, + "Failed to broadcast transaction {}: {:?}", + txid, + e + ); + } + } + }, + } + } + + /// Reconciles the block-fee cache against the canonical chain and returns the per-block fee + /// rates for the most recent [`FEE_WINDOW_BLOCKS`] blocks. + /// + /// For each height in the window we fetch the canonical block hash; if the cached entry still + /// matches we reuse its rate, otherwise (new block, or a block that was reorged out) we download + /// it via [`Requester::average_fee_rate`]. Heights outside the window are evicted by replacing + /// the cache with the freshly built window. + /// + /// This is best-effort: a height we can't fetch a header or block for is simply skipped (so a + /// slow or unresponsive peer can't stall or void the whole update), and an empty result just + /// means we have no recent data yet. The window therefore fills incrementally over successive + /// updates rather than requiring all [`FEE_WINDOW_BLOCKS`] downloads to succeed at once. + async fn refresh_block_fee_window( + &self, requester: &Requester, cache: &Mutex>, + ) -> Vec { + let tip_height = match requester.chain_tip().await { + Ok(tip) => tip.height, + Err(e) => { + log_error!(self.logger, "CBF fee update: failed to fetch chain tip: {:?}", e); + return Vec::new(); + }, + }; + let lo = tip_height.saturating_sub(FEE_WINDOW_BLOCKS - 1); + + // Snapshot the cache so we never hold the std `Mutex` across an `.await`. + let cached = cache.lock().expect("lock").clone(); + + let mut window = BTreeMap::new(); + for height in lo..=tip_height { + let canonical_hash = match requester.get_header(height).await { + // Height not available (yet); skip it. + Ok(None) => continue, + Ok(Some(header)) => header.block_hash(), + Err(e) => { + log_debug!( + self.logger, + "CBF fee update: failed to fetch header at height {}, skipping: {:?}", + height, + e + ); + continue; + }, + }; + + // Reuse the cached rate while the block is still canonical; otherwise download it. + if let Some((hash, fee_rate)) = cached.get(&height) { + if *hash == canonical_hash { + window.insert(height, (canonical_hash, *fee_rate)); + continue; + } + } + + match tokio::time::timeout( + Duration::from_secs(CBF_BLOCK_FETCH_TIMEOUT_SECS), + requester.average_fee_rate(canonical_hash), + ) + .await + { + Ok(Ok(fee_rate)) => { + window.insert(height, (canonical_hash, fee_rate)); + }, + Ok(Err(e)) => { + log_debug!( + self.logger, + "CBF fee update: failed to fetch fee rate for block {}, skipping: {:?}", + canonical_hash, + e + ); + }, + Err(_) => { + log_debug!( + self.logger, + "CBF fee update: timed out fetching block {} for fee estimation, skipping.", + canonical_hash, + ); + }, + } + } + + let samples = window.values().map(|(_, fee_rate)| *fee_rate).collect(); + // Replacing the cache wholesale also evicts any entries that fell out of the window. + *cache.lock().expect("lock") = window; + samples + } +} + +fn resume_checkpoint(logger: &Logger, chain_listener: &ChainListener) -> Option { + let min_best_block = chain_listener.get_best_block(); + let bdk_cp = chain_listener.onchain_wallet.latest_checkpoint(); + + if let Some(bdk_at_height) = bdk_cp.get(min_best_block.height) { + if bdk_at_height.hash() != min_best_block.block_hash { + log_error!( + logger, + "CBF resume: listener best block at height {} has hash {} but BDK has {}; \ + a component may be on a stale fork. Anchoring on BDK's chain.", + min_best_block.height, + min_best_block.block_hash, + bdk_at_height.hash(), + ); + } + } + + // Walk BDK's checkpoint chain back to the reorg-safe anchor height. + let target_height = min_best_block.height.saturating_sub(REORG_SAFETY_BLOCKS); + let mut cursor = bdk_cp; + while cursor.height() > target_height { + match cursor.prev() { + Some(prev) => cursor = prev, + None => break, + } + } + + (cursor.height() > 0).then(|| HashCheckpoint::new(cursor.height(), cursor.hash())) +} diff --git a/src/chain/electrum.rs b/src/chain/electrum.rs index 59fa23a6c..c1d04dc6f 100644 --- a/src/chain/electrum.rs +++ b/src/chain/electrum.rs @@ -289,7 +289,14 @@ impl ElectrumChainSource { let now = Instant::now(); - let new_fee_rate_cache = electrum_client.get_fee_rate_cache_update().await?; + let new_fee_rate_cache = get_electrum_fee_rate_cache_update( + Arc::clone(&electrum_client.runtime), + Arc::clone(&electrum_client.electrum_client), + self.config.network, + self.sync_config.timeouts_config.fee_rate_cache_update_timeout_secs, + Arc::clone(&self.logger), + ) + .await?; self.fee_estimator.set_fee_rate_cache(new_fee_rate_cache); log_debug!( @@ -708,91 +715,84 @@ impl ElectrumRuntimeClient { Err(e) => self.log_broadcast_error(e, &txids, &package), } } +} - async fn get_fee_rate_cache_update( - &self, - ) -> Result, Error> { - let electrum_client = Arc::clone(&self.electrum_client); - - let mut batch = Batch::default(); - let confirmation_targets = get_all_conf_targets(); - for target in confirmation_targets { - let num_blocks = get_num_block_defaults_for_target(target); - batch.estimate_fee(num_blocks, None); - } - - let spawn_fut = self.runtime.spawn_blocking(move || electrum_client.batch_call(&batch)); +pub(crate) async fn get_electrum_fee_rate_cache_update( + runtime: Arc, electrum_client: Arc, network: Network, + fee_rate_cache_update_timeout_secs: u64, logger: Arc, +) -> Result, Error> { + let mut batch = Batch::default(); + let confirmation_targets = get_all_conf_targets(); + for target in confirmation_targets { + let num_blocks = get_num_block_defaults_for_target(target); + batch.estimate_fee(num_blocks, None); + } - let timeout_fut = tokio::time::timeout( - Duration::from_secs( - self.sync_config.timeouts_config.fee_rate_cache_update_timeout_secs, - ), - spawn_fut, + let spawn_fut = runtime.spawn_blocking(move || electrum_client.batch_call(&batch)); + + let timeout_fut = + tokio::time::timeout(Duration::from_secs(fee_rate_cache_update_timeout_secs), spawn_fut); + + let raw_estimates_btc_kvb = timeout_fut + .await + .map_err(|e| { + log_error!(logger, "Updating fee rate estimates timed out: {}", e); + Error::FeerateEstimationUpdateTimeout + })? + .map_err(|e| { + log_error!(logger, "Failed to retrieve fee rate estimates: {}", e); + Error::FeerateEstimationUpdateFailed + })? + .map_err(|e| { + log_error!(logger, "Failed to retrieve fee rate estimates: {}", e); + Error::FeerateEstimationUpdateFailed + })?; + + if raw_estimates_btc_kvb.len() != confirmation_targets.len() && network == Network::Bitcoin { + // Ensure we fail if we didn't receive all estimates. + debug_assert!( + false, + "Electrum server didn't return all expected results. This is disallowed on Mainnet." ); - - let raw_estimates_btc_kvb = timeout_fut - .await - .map_err(|e| { - log_error!(self.logger, "Updating fee rate estimates timed out: {}", e); - Error::FeerateEstimationUpdateTimeout - })? - .map_err(|e| { - log_error!(self.logger, "Failed to retrieve fee rate estimates: {}", e); - Error::FeerateEstimationUpdateFailed - })? - .map_err(|e| { - log_error!(self.logger, "Failed to retrieve fee rate estimates: {}", e); - Error::FeerateEstimationUpdateFailed - })?; - - if raw_estimates_btc_kvb.len() != confirmation_targets.len() - && self.config.network == Network::Bitcoin - { - // Ensure we fail if we didn't receive all estimates. - debug_assert!(false, - "Electrum server didn't return all expected results. This is disallowed on Mainnet." - ); - log_error!(self.logger, + log_error!(logger, "Failed to retrieve fee rate estimates: Electrum server didn't return all expected results. This is disallowed on Mainnet." ); - return Err(Error::FeerateEstimationUpdateFailed); - } + return Err(Error::FeerateEstimationUpdateFailed); + } - let mut new_fee_rate_cache = HashMap::with_capacity(10); - for (target, raw_fee_rate_btc_per_kvb) in - confirmation_targets.into_iter().zip(raw_estimates_btc_kvb.into_iter()) - { - // Parse the retrieved serde_json::Value and fall back to 1 sat/vb (10^3 / 10^8 = 10^-5 - // = 0.00001 btc/kvb) if we fail or it yields less than that. This is mostly necessary - // to continue on `signet`/`regtest` where we might not get estimates (or bogus - // values). - let fee_rate_btc_per_kvb = raw_fee_rate_btc_per_kvb - .as_f64() - .map_or(0.00001, |converted| converted.max(0.00001)); - - // Electrum, just like Bitcoin Core, gives us a feerate in BTC/KvB. - // Thus, we multiply by 25_000_000 (10^8 / 4) to get satoshis/kwu. - let fee_rate = { - let fee_rate_sat_per_kwu = (fee_rate_btc_per_kvb * 25_000_000.0).round() as u64; - FeeRate::from_sat_per_kwu(fee_rate_sat_per_kwu) - }; - - // LDK 0.0.118 introduced changes to the `ConfirmationTarget` semantics that - // require some post-estimation adjustments to the fee rates, which we do here. - let adjusted_fee_rate = apply_post_estimation_adjustments(target, fee_rate); - - new_fee_rate_cache.insert(target, adjusted_fee_rate); - - log_trace!( - self.logger, - "Fee rate estimation updated for {:?}: {} sats/kwu", - target, - adjusted_fee_rate.to_sat_per_kwu(), - ); - } + let mut new_fee_rate_cache = HashMap::with_capacity(10); + for (target, raw_fee_rate_btc_per_kvb) in + confirmation_targets.into_iter().zip(raw_estimates_btc_kvb.into_iter()) + { + // Parse the retrieved serde_json::Value and fall back to 1 sat/vb (10^3 / 10^8 = 10^-5 + // = 0.00001 btc/kvb) if we fail or it yields less than that. This is mostly necessary + // to continue on `signet`/`regtest` where we might not get estimates (or bogus + // values). + let fee_rate_btc_per_kvb = + raw_fee_rate_btc_per_kvb.as_f64().map_or(0.00001, |converted| converted.max(0.00001)); + + // Electrum, just like Bitcoin Core, gives us a feerate in BTC/KvB. + // Thus, we multiply by 25_000_000 (10^8 / 4) to get satoshis/kwu. + let fee_rate = { + let fee_rate_sat_per_kwu = (fee_rate_btc_per_kvb * 25_000_000.0).round() as u64; + FeeRate::from_sat_per_kwu(fee_rate_sat_per_kwu) + }; + + // LDK 0.0.118 introduced changes to the `ConfirmationTarget` semantics that + // require some post-estimation adjustments to the fee rates, which we do here. + let adjusted_fee_rate = apply_post_estimation_adjustments(target, fee_rate); - Ok(new_fee_rate_cache) + new_fee_rate_cache.insert(target, adjusted_fee_rate); + + log_trace!( + logger, + "Fee rate estimation updated for {:?}: {} sats/kwu", + target, + adjusted_fee_rate.to_sat_per_kwu(), + ); } + + Ok(new_fee_rate_cache) } impl Filter for ElectrumRuntimeClient { diff --git a/src/chain/mod.rs b/src/chain/mod.rs index 0f96c409f..f4673da2e 100644 --- a/src/chain/mod.rs +++ b/src/chain/mod.rs @@ -6,6 +6,7 @@ // accordance with one or both of these licenses. pub(crate) mod bitcoind; +mod cbf; mod electrum; mod esplora; @@ -16,7 +17,8 @@ use std::time::Duration; use bitcoin::{Script, Txid}; use lightning::chain::{BlockLocator, Filter}; -use crate::chain::bitcoind::{BitcoindChainSource, UtxoSourceClient}; +use crate::chain::bitcoind::{BitcoindChainSource, ChainListener, UtxoSourceClient}; +use crate::chain::cbf::CbfChainSource; use crate::chain::electrum::ElectrumChainSource; use crate::chain::esplora::EsploraChainSource; use crate::config::{ @@ -113,6 +115,20 @@ impl WalletSyncStatus { } } +/// Optional external fee estimation backend for the CBF chain source. +/// +/// By default CBF derives fee rates from recent blocks' coinbase outputs. +/// Setting an external source provides more accurate, per-target estimates +/// from a mempool-aware server. +#[derive(Debug, Clone)] +#[cfg_attr(feature = "uniffi", derive(uniffi::Enum))] +pub enum CbfFeeSourceConfig { + /// Use an Esplora HTTP server for fee rate estimation. + Esplora(String), + /// Use an Electrum server for fee rate estimation. + Electrum(String), +} + pub(crate) struct ChainSource { kind: ChainSourceKind, registered_txids: Mutex>, @@ -124,6 +140,7 @@ enum ChainSourceKind { Esplora(EsploraChainSource), Electrum(ElectrumChainSource), Bitcoind(BitcoindChainSource), + Cbf(CbfChainSource), } impl ChainSource { @@ -215,11 +232,45 @@ impl ChainSource { (Self { kind, registered_txids, tx_broadcaster, logger }, best_block) } - pub(crate) fn start(&self, runtime: Arc) -> Result<(), Error> { + pub(crate) fn new_cbf( + peers: Vec, fee_source_config: Option, runtime: Arc, + fee_estimator: Arc, tx_broadcaster: Arc, + kv_store: Arc, config: Arc, logger: Arc, + node_metrics: Arc, + ) -> Result<(Self, Option), Error> { + let cbf_chain_source = CbfChainSource::new( + peers, + fee_source_config, + runtime, + Arc::clone(&fee_estimator), + Arc::clone(&kv_store), + Arc::clone(&config), + Arc::clone(&logger), + Arc::clone(&node_metrics), + )?; + let kind = ChainSourceKind::Cbf(cbf_chain_source); + let registered_txids = Mutex::new(HashSet::new()); + Ok((Self { kind, registered_txids, tx_broadcaster, logger }, None)) + } + + pub(crate) fn start( + &self, runtime: Arc, onchain_wallet: Arc, + channel_manager: Arc, chain_monitor: Arc, + output_sweeper: Arc, + ) -> Result<(), Error> { match &self.kind { ChainSourceKind::Electrum(electrum_chain_source) => { electrum_chain_source.start(runtime)? }, + ChainSourceKind::Cbf(cbf_chain_source) => { + let chain_listener = ChainListener { + onchain_wallet, + channel_manager, + chain_monitor, + output_sweeper, + }; + cbf_chain_source.start(chain_listener); + }, _ => { // Nothing to do for other chain sources. }, @@ -230,6 +281,7 @@ impl ChainSource { pub(crate) fn stop(&self) { match &self.kind { ChainSourceKind::Electrum(electrum_chain_source) => electrum_chain_source.stop(), + ChainSourceKind::Cbf(cbf_chain_source) => cbf_chain_source.stop(), _ => { // Nothing to do for other chain sources. }, @@ -254,6 +306,7 @@ impl ChainSource { ChainSourceKind::Esplora(_) => true, ChainSourceKind::Electrum { .. } => true, ChainSourceKind::Bitcoind { .. } => false, + ChainSourceKind::Cbf { .. } => false, } } @@ -280,9 +333,9 @@ impl ChainSource { } else { // Background syncing is disabled log_info!( - self.logger, - "Background syncing is disabled. Manual syncing required for onchain wallet, lightning wallet, and fee rate updates.", - ); + self.logger, + "Background syncing is disabled. Manual syncing required for onchain wallet, lightning wallet, and fee rate updates.", + ); return; } }, @@ -303,9 +356,9 @@ impl ChainSource { } else { // Background syncing is disabled log_info!( - self.logger, - "Background syncing is disabled. Manual syncing required for onchain wallet, lightning wallet, and fee rate updates.", - ); + self.logger, + "Background syncing is disabled. Manual syncing required for onchain wallet, lightning wallet, and fee rate updates.", + ); return; } }, @@ -320,6 +373,11 @@ impl ChainSource { ) .await }, + ChainSourceKind::Cbf(cbf_chain_source) => { + //CBF cannot run without background syncing, when the chain source is running, it + //syncs. Thus we don't have anything similar to other chain sources. + cbf_chain_source.continuously_update_fee_rate_estimates(stop_sync_receiver).await + }, } } @@ -362,7 +420,7 @@ impl ChainSource { log_trace!( logger, "Stopping background syncing on-chain wallet.", - ); + ); return; } _ = onchain_wallet_sync_interval.tick() => { @@ -376,7 +434,7 @@ impl ChainSource { Arc::clone(&channel_manager), Arc::clone(&chain_monitor), Arc::clone(&output_sweeper), - ).await; + ).await; } } } @@ -399,6 +457,9 @@ impl ChainSource { // `ChainPoller`. So nothing to do here. unreachable!("Onchain wallet will be synced via chain polling") }, + ChainSourceKind::Cbf { .. } => { + unreachable!("Onchain wallet synchronizes in background") + }, } } @@ -424,6 +485,9 @@ impl ChainSource { // `ChainPoller`. So nothing to do here. unreachable!("Lightning wallet will be synced via chain polling") }, + ChainSourceKind::Cbf { .. } => { + unreachable!("Lightning wallet synchronizes in background") + }, } } @@ -452,6 +516,7 @@ impl ChainSource { ) .await }, + ChainSourceKind::Cbf(cbf_chain_source) => cbf_chain_source.wait_until_synced().await, } } @@ -466,6 +531,9 @@ impl ChainSource { ChainSourceKind::Bitcoind(bitcoind_chain_source) => { bitcoind_chain_source.update_fee_rate_estimates().await }, + ChainSourceKind::Cbf(cbf_chain_source) => { + cbf_chain_source.update_fee_rate_estimates().await + }, } } @@ -486,6 +554,13 @@ impl ChainSource { ChainSourceKind::Bitcoind(bitcoind_chain_source) => { bitcoind_chain_source.validate_zero_fee_commitments_support().await }, + ChainSourceKind::Cbf(_) => { + log_error!( + self.logger, + "CBF chain sources cannot verify zero-fee commitment package relay support" + ); + Err(Error::ChainSourceNotSupported) + }, } } @@ -529,6 +604,11 @@ impl ChainSource { ChainSourceKind::Bitcoind(bitcoind_chain_source) => { bitcoind_chain_source.process_transaction_broadcast(package).await }, + ChainSourceKind::Cbf(cbf_chain_source) => { + cbf_chain_source + .process_broadcast_package(package.into_inner()) + .await + }, } } } @@ -547,6 +627,9 @@ impl Filter for ChainSource { electrum_chain_source.register_tx(txid, script_pubkey) }, ChainSourceKind::Bitcoind { .. } => (), + ChainSourceKind::Cbf(cbf_chain_source) => { + cbf_chain_source.register_tx(txid, script_pubkey); + }, } } fn register_output(&self, output: lightning::chain::WatchedOutput) { @@ -558,6 +641,9 @@ impl Filter for ChainSource { electrum_chain_source.register_output(output) }, ChainSourceKind::Bitcoind { .. } => (), + ChainSourceKind::Cbf(cbf_chain_source) => { + cbf_chain_source.register_output(output); + }, } } } diff --git a/src/config.rs b/src/config.rs index 958eb14fc..d4ac48093 100644 --- a/src/config.rs +++ b/src/config.rs @@ -26,7 +26,7 @@ use crate::logger::LogLevel; const DEFAULT_NETWORK: Network = Network::Bitcoin; const DEFAULT_BDK_WALLET_SYNC_INTERVAL_SECS: u64 = 80; const DEFAULT_LDK_WALLET_SYNC_INTERVAL_SECS: u64 = 30; -const DEFAULT_FEE_RATE_CACHE_UPDATE_INTERVAL_SECS: u64 = 60 * 10; +pub(crate) const DEFAULT_FEE_RATE_CACHE_UPDATE_INTERVAL_SECS: u64 = 60 * 10; const DEFAULT_PROBING_LIQUIDITY_LIMIT_MULTIPLIER: u64 = 3; pub(crate) const DEFAULT_PROBING_INTERVAL_SECS: u64 = 10; pub(crate) const MIN_PROBING_INTERVAL: Duration = Duration::from_millis(100); diff --git a/src/lib.rs b/src/lib.rs index cb570243c..06c782d82 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -300,11 +300,19 @@ impl Node { self.runtime.allow_cancellable_background_task_spawns(); - // Start up any runtime-dependant chain sources (e.g. Electrum) - self.chain_source.start(Arc::clone(&self.runtime)).map_err(|e| { - log_error!(self.logger, "Failed to start chain syncing: {}", e); - e - })?; + // Start up any runtime-dependant chain sources (e.g. Electrum, CBF) + self.chain_source + .start( + Arc::clone(&self.runtime), + Arc::clone(&self.wallet), + Arc::clone(&self.channel_manager), + Arc::clone(&self.chain_monitor), + Arc::clone(&self.output_sweeper), + ) + .map_err(|e| { + log_error!(self.logger, "Failed to start chain syncing: {}", e); + e + })?; let manager_owns_any_0fc_channels = self.channel_manager.list_channels().into_iter().any(|channel| { @@ -849,13 +857,15 @@ impl Node { self.peer_manager.disconnect_all_peers(); log_debug!(self.logger, "Disconnected all network peers."); - // Wait until non-cancellable background tasks (mod LDK's background processor) are done. - self.runtime.wait_on_background_tasks(); - - // Stop any runtime-dependant chain sources. + // Stop any runtime-dependant chain sources before waiting on non-cancellable + // background tasks. Some chain sources own background tasks that only exit + // after their client/requester is shut down. self.chain_source.stop(); log_debug!(self.logger, "Stopped chain sources."); + // Wait until non-cancellable background tasks (mod LDK's background processor) are done. + self.runtime.wait_on_background_tasks(); + // Stop the background processor. self.background_processor_stop_sender .send(()) @@ -1987,6 +1997,9 @@ impl Node { /// However, if background syncing is disabled (i.e., `background_sync_config` is set to `None`), /// this method must be called manually to keep wallets in sync with the chain state. /// + /// When using the CBF chain source, syncing always runs in the background. In that mode this + /// method waits until the background sync has applied chain updates through the current tip. + /// /// [`EsploraSyncConfig::background_sync_config`]: crate::config::EsploraSyncConfig::background_sync_config pub fn sync_wallets(&self) -> Result<(), Error> { if !*self.is_running.read().expect("lock") { diff --git a/src/util.rs b/src/util.rs index 3350ad2c7..8cd86665a 100644 --- a/src/util.rs +++ b/src/util.rs @@ -5,6 +5,61 @@ // http://opensource.org/licenses/MIT>, at your option. You may not use this file except in // accordance with one or both of these licenses. +//! Miscellaneous pure helper functions. + +use bitcoin::constants::SUBSIDY_HALVING_INTERVAL; +use bitcoin::{Amount, Block, FeeRate}; + +use crate::fee_estimator::{get_num_block_defaults_for_target, ConfirmationTarget}; + +/// Block subsidy at the given height (approximate on regtest). +pub(crate) fn block_subsidy(height: u32) -> Amount { + let halvings = height / SUBSIDY_HALVING_INTERVAL; + if halvings >= 64 { + return Amount::ZERO; + } + Amount::from_sat((Amount::ONE_BTC.to_sat() * 50) >> halvings) +} + +/// Average fee rate of a block, derived from its coinbase: `(coinbase output total - subsidy) / +/// weight`. Lets us compute the fee rate of a block we already hold without a re-download. +pub(crate) fn coinbase_fee_rate(block: &Block, height: u32) -> FeeRate { + let revenue: Amount = block + .txdata + .first() + .map(|coinbase| coinbase.output.iter().map(|txout| txout.value).sum()) + .unwrap_or(Amount::ZERO); + let block_fees = revenue.checked_sub(block_subsidy(height)).unwrap_or(Amount::ZERO); + let fee_rate = block_fees.to_sat().checked_div(block.weight().to_kwu_floor()).unwrap_or(0); + FeeRate::from_sat_per_kwu(fee_rate) +} + +/// Maps a confirmation target to the percentile of the recent-block fee-rate window we read for it. +/// +/// More urgent targets (shorter confirmation horizon) read a higher percentile; relaxed targets +/// read a lower one. This is a coarse stand-in for the per-horizon estimates a mempool-aware +/// backend would provide. +pub(crate) fn cbf_percentile_for_target(target: ConfirmationTarget) -> f64 { + match get_num_block_defaults_for_target(target) { + 0..=2 => 90.0, + 3..=6 => 75.0, + 7..=12 => 50.0, + 13..=144 => 25.0, + _ => 10.0, + } +} + +/// Returns the value at the given percentile of an ascending-sorted slice using nearest-rank. +/// Returns `0` for an empty slice. +pub(crate) fn percentile_of_sorted(sorted: &[u64], percentile: f64) -> u64 { + if sorted.is_empty() { + return 0; + } + let rank = ((percentile / 100.0) * sorted.len() as f64).ceil() as usize; + let idx = rank.saturating_sub(1).min(sorted.len() - 1); + sorted[idx] +} + /// Returns a random `u64` uniformly distributed in `[min, max]` (inclusive). pub(crate) fn random_range(min: u64, max: u64) -> u64 { debug_assert!(min <= max); diff --git a/src/wallet/mod.rs b/src/wallet/mod.rs index f8d9d521e..690ab7caa 100644 --- a/src/wallet/mod.rs +++ b/src/wallet/mod.rs @@ -141,6 +141,10 @@ impl Wallet { .collect() } + pub(crate) fn latest_checkpoint(&self) -> bdk_chain::local_chain::CheckPoint { + self.inner.lock().expect("lock").latest_checkpoint() + } + pub(crate) fn current_best_block(&self) -> BlockLocator { let checkpoint = self.inner.lock().expect("lock").latest_checkpoint(); let mut current_block = Some(checkpoint.clone()); @@ -212,6 +216,14 @@ impl Wallet { Ok(()) } + /// Returns every script pubkey the wallet is watching for on-chain activity: all revealed + /// SPKs plus the lookahead window BDK derives beyond the last revealed index on each keychain. + /// A block may pay an address we have not explicitly revealed yet (e.g. on recovery, where a fresh + /// wallet has revealed nothing) but which is still within the gap limit. + pub(crate) fn list_watched_scripts(&self) -> Vec { + self.inner.lock().expect("lock").spk_index().inner().all_spks().values().cloned().collect() + } + async fn update_payment_store(&self, mut events: Vec) -> Result<(), Error> { if events.is_empty() { return Ok(()); @@ -1842,13 +1854,14 @@ fn aggregate_local_stakes(candidate: &FundingCandidate) -> LocalStakeAggregate { impl Listen for Wallet { fn filtered_block_connected( - &self, _header: &bitcoin::block::Header, - _txdata: &lightning::chain::transaction::TransactionData, _height: u32, + &self, header: &bitcoin::block::Header, + _txdata: &lightning::chain::transaction::TransactionData, height: u32, ) { - debug_assert!(false, "Syncing filtered blocks is currently not supported"); - // As far as we can tell this would be a no-op anyways as we don't have to tell BDK about - // the header chain of intermediate blocks. According to the BDK team, it's sufficient to - // only connect full blocks starting from the last point of disagreement. + // A non-matching filter means none of this block's transactions are relevant to us, so there + // is nothing but the header to apply. We still connect an empty block built from the header + // to keep the on-chain wallet's chain contiguous with the listeners. + let block = bitcoin::Block { header: *header, txdata: Vec::new() }; + self.block_connected(&block, height); } fn block_connected(&self, block: &bitcoin::Block, height: u32) { diff --git a/tests/common/mod.rs b/tests/common/mod.rs index 50e2b993c..ab911c697 100644 --- a/tests/common/mod.rs +++ b/tests/common/mod.rs @@ -314,6 +314,10 @@ pub(crate) fn setup_bitcoind_and_electrsd() -> (BitcoinD, ElectrsD) { let mut bitcoind_conf = corepc_node::Conf::default(); bitcoind_conf.network = "regtest"; bitcoind_conf.args.push("-rest"); + // Enable P2P and compact block filters so the CBF (BIP157) chain source can connect and sync. + bitcoind_conf.p2p = corepc_node::P2P::Yes; + bitcoind_conf.args.push("-blockfilterindex=1"); + bitcoind_conf.args.push("-peerblockfilters=1"); let bitcoind = BitcoinD::with_conf(bitcoind_exe, &bitcoind_conf).unwrap(); let electrs_exe = env::var("ELECTRS_EXE") @@ -330,7 +334,14 @@ pub(crate) fn setup_bitcoind_and_electrsd() -> (BitcoinD, ElectrsD) { pub(crate) fn random_chain_source<'a>( bitcoind: &'a BitcoinD, electrsd: &'a ElectrsD, ) -> TestChainSource<'a> { - let r = rand::random_range(0..3); + let r = match std::env::var("LDK_TEST_CHAIN_SOURCE").ok().as_deref() { + Some("esplora") => 0, + Some("electrum") => 1, + Some("bitcoind-rpc") => 2, + Some("bitcoind-rest") => 3, + Some("cbf") => 4, + _ => rand::random_range(0..3), + }; match r { 0 => { println!("Randomly setting up Esplora chain syncing..."); @@ -348,6 +359,10 @@ pub(crate) fn random_chain_source<'a>( println!("Randomly setting up Bitcoind REST chain syncing..."); TestChainSource::BitcoindRestSync(bitcoind) }, + 4 => { + println!("Randomly setting up CBF compact block filter syncing..."); + TestChainSource::Cbf(bitcoind) + }, _ => unreachable!(), } } @@ -475,7 +490,10 @@ async fn settle_force_close_balance( assert_eq!(actual_counterparty_node_id, counterparty_node_id); let cur_height = node.status().current_best_block.height; let blocks_to_go = confirmation_height - cur_height; - generate_blocks_and_wait(bitcoind, electrsd, blocks_to_go as usize).await; + let new_height = + generate_blocks_and_wait(bitcoind, electrsd, blocks_to_go as usize).await; + wait_for_node_tip(node, new_height).await; + wait_for_node_tip(peer_node, new_height).await; node.sync_wallets().unwrap(); peer_node.sync_wallets().unwrap(); }, @@ -490,7 +508,9 @@ async fn settle_force_close_balance( if node.list_balances().lightning_balances.is_empty() { break; } - generate_blocks_and_wait(bitcoind, electrsd, 1).await; + let new_height = generate_blocks_and_wait(bitcoind, electrsd, 1).await; + wait_for_node_tip(node, new_height).await; + wait_for_node_tip(peer_node, new_height).await; node.sync_wallets().unwrap(); peer_node.sync_wallets().unwrap(); } @@ -500,7 +520,9 @@ async fn settle_force_close_balance( assert_eq!(balances.pending_balances_from_channel_closures.len(), 1); match balances.pending_balances_from_channel_closures[0] { PendingSweepBalance::BroadcastAwaitingConfirmation { .. } => { - generate_blocks_and_wait(bitcoind, electrsd, 1).await; + let new_height = generate_blocks_and_wait(bitcoind, electrsd, 1).await; + wait_for_node_tip(node, new_height).await; + wait_for_node_tip(peer_node, new_height).await; node.sync_wallets().unwrap(); peer_node.sync_wallets().unwrap(); @@ -515,7 +537,9 @@ async fn settle_force_close_balance( _ => panic!("Unexpected balance state!"), } - generate_blocks_and_wait(bitcoind, electrsd, 5).await; + let new_height = generate_blocks_and_wait(bitcoind, electrsd, 5).await; + wait_for_node_tip(node, new_height).await; + wait_for_node_tip(peer_node, new_height).await; node.sync_wallets().unwrap(); peer_node.sync_wallets().unwrap(); } @@ -526,6 +550,7 @@ pub(crate) enum TestChainSource<'a> { Electrum(&'a ElectrsD), BitcoindRpcSync(&'a BitcoinD), BitcoindRestSync(&'a BitcoinD), + Cbf(&'a BitcoinD), } #[derive(Clone, Copy)] @@ -698,6 +723,11 @@ pub(crate) fn setup_node(chain_source: &TestChainSource, config: TestConfig) -> config.wallet_rescan_from_height, ); }, + TestChainSource::Cbf(bitcoind) => { + let p2p_socket = bitcoind.params.p2p_socket.expect("P2P must be enabled for CBF"); + let peer_addr = format!("{}", p2p_socket); + builder.set_chain_source_cbf(vec![peer_addr], None); + }, } match &config.log_writer { @@ -737,7 +767,7 @@ pub(crate) fn setup_node(chain_source: &TestChainSource, config: TestConfig) -> pub(crate) async fn generate_blocks_and_wait( bitcoind: &BitcoindClient, electrs: &E, num: usize, -) { +) -> usize { let _ = bitcoind.create_wallet("ldk_node_test"); let _ = bitcoind.load_wallet("ldk_node_test"); print!("Generating {} blocks...", num); @@ -746,9 +776,11 @@ pub(crate) async fn generate_blocks_and_wait( let address = bitcoind.new_address().expect("failed to get new address"); // TODO: expect this Result once the WouldBlock issue is resolved upstream. let _block_hashes_res = bitcoind.generate_to_address(num, &address); - wait_for_block(bitcoind, electrs, cur_height as usize + num).await; + let new_height = cur_height as usize + num; + wait_for_block(bitcoind, electrs, new_height).await; print!(" Done!"); println!("\n"); + return new_height; } pub(crate) fn invalidate_blocks(bitcoind: &BitcoindClient, num_blocks: usize) { @@ -847,6 +879,13 @@ pub(crate) async fn wait_for_channel_ready_to_send( ); } +pub(crate) async fn wait_for_node_tip(node: &Node, height: usize) { + exponential_backoff_poll(|| { + (node.status().current_best_block.height as usize >= height).then_some(()) + }) + .await; +} + pub(crate) async fn exponential_backoff_poll(mut poll: F) -> T where F: FnMut() -> Option, @@ -1156,7 +1195,9 @@ pub(crate) async fn do_channel_full_cycle( wait_for_tx(electrsd, funding_txo_a.txid).await; if !allow_0conf { - generate_blocks_and_wait(&bitcoind, electrsd, 6).await; + let new_height = generate_blocks_and_wait(&bitcoind, electrsd, 6).await; + wait_for_node_tip(&node_a, new_height).await; + wait_for_node_tip(&node_b, new_height).await; } node_a.sync_wallets().unwrap(); @@ -1528,7 +1569,9 @@ pub(crate) async fn do_channel_full_cycle( ); // Mine a block to give time for the HTLC to resolve - generate_blocks_and_wait(&bitcoind, electrsd, 1).await; + let new_height = generate_blocks_and_wait(&bitcoind, electrsd, 1).await; + wait_for_node_tip(&node_a, new_height).await; + wait_for_node_tip(&node_b, new_height).await; println!("\nB splices out to pay A"); let addr_a = node_a.onchain_payment().new_address().unwrap(); @@ -1540,7 +1583,11 @@ pub(crate) async fn do_channel_full_cycle( expect_splice_negotiated_event!(node_a, node_b.node_id()); expect_splice_negotiated_event!(node_b, node_a.node_id()); - generate_blocks_and_wait(&bitcoind, electrsd, 6).await; + tokio::time::sleep(Duration::from_secs(2)).await; + + let new_height = generate_blocks_and_wait(&bitcoind, electrsd, 6).await; + wait_for_node_tip(&node_a, new_height).await; + wait_for_node_tip(&node_b, new_height).await; node_a.sync_wallets().unwrap(); node_b.sync_wallets().unwrap(); @@ -1562,7 +1609,10 @@ pub(crate) async fn do_channel_full_cycle( expect_splice_negotiated_event!(node_a, node_b.node_id()); expect_splice_negotiated_event!(node_b, node_a.node_id()); - generate_blocks_and_wait(&bitcoind, electrsd, 6).await; + tokio::time::sleep(Duration::from_secs(5)).await; + let new_height = generate_blocks_and_wait(&bitcoind, electrsd, 6).await; + wait_for_node_tip(&node_a, new_height).await; + wait_for_node_tip(&node_b, new_height).await; node_a.sync_wallets().unwrap(); node_b.sync_wallets().unwrap(); @@ -1612,8 +1662,10 @@ pub(crate) async fn do_channel_full_cycle( tokio::time::sleep(Duration::from_secs(1)).await; if force_close { node_a.force_close_channel(&user_channel_id_a, node_b.node_id(), None).unwrap(); + tokio::time::sleep(Duration::from_secs(2)).await; } else { node_a.close_channel(&user_channel_id_a, node_b.node_id()).unwrap(); + tokio::time::sleep(Duration::from_secs(2)).await; // The cooperative shutdown may complete before we get to check, but if the channel // is still visible it must already be in a shutdown state. if let Some(channel) = @@ -1635,7 +1687,9 @@ pub(crate) async fn do_channel_full_cycle( wait_for_outpoint_spend(electrsd, funding_txo_b).await; - generate_blocks_and_wait(&bitcoind, electrsd, 1).await; + let new_height = generate_blocks_and_wait(&bitcoind, electrsd, 1).await; + wait_for_node_tip(&node_a, new_height).await; + wait_for_node_tip(&node_b, new_height).await; node_a.sync_wallets().unwrap(); node_b.sync_wallets().unwrap(); @@ -1680,7 +1734,10 @@ pub(crate) async fn do_channel_full_cycle( assert_eq!(node_a_blocks_to_go, node_b_blocks_to_go); - generate_blocks_and_wait(&bitcoind, electrsd, node_a_blocks_to_go as usize).await; + let new_height = + generate_blocks_and_wait(&bitcoind, electrsd, node_a_blocks_to_go as usize).await; + wait_for_node_tip(&node_a, new_height).await; + wait_for_node_tip(&node_b, new_height).await; node_a.sync_wallets().unwrap(); node_b.sync_wallets().unwrap(); diff --git a/tests/integration_tests_rust.rs b/tests/integration_tests_rust.rs index ba15f6fa3..9e762d9c6 100644 --- a/tests/integration_tests_rust.rs +++ b/tests/integration_tests_rust.rs @@ -27,8 +27,8 @@ use common::{ generate_listening_addresses, invalidate_blocks, open_channel, open_channel_push_amt, open_channel_with_all, premine_and_distribute_funds, premine_blocks, prepare_rbf, random_chain_source, random_config, setup_bitcoind_and_electrsd, setup_builder, setup_node, - setup_two_nodes, splice_in_with_all, wait_for_block, wait_for_tx, InMemoryStore, - TestChainSource, TestConfig, TestStoreType, TestSyncStore, + setup_two_nodes, splice_in_with_all, wait_for_block, wait_for_node_tip, wait_for_tx, + InMemoryStore, TestChainSource, TestConfig, TestStoreType, TestSyncStore, }; use electrsd::corepc_node::{self, Node as BitcoinD}; use electrsd::ElectrsD; @@ -724,10 +724,11 @@ async fn onchain_send_receive() { let channel_amount_sat = 1_000_000; let reserve_amount_sat = 25_000; open_channel(&node_b, &node_a, channel_amount_sat, true, &electrsd).await; - generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 6).await; - + let new_height = generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 6).await; node_a.sync_wallets().unwrap(); node_b.sync_wallets().unwrap(); + wait_for_node_tip(&node_a, new_height).await; + wait_for_node_tip(&node_b, new_height).await; expect_channel_ready_event!(node_a, node_b.node_id()); expect_channel_ready_event!(node_b, node_a.node_id()); @@ -795,9 +796,11 @@ async fn onchain_send_receive() { assert_eq!(payment_a.amount_msat, payment_b.amount_msat); assert_eq!(payment_a.fee_paid_msat, payment_b.fee_paid_msat); - generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 6).await; + let new_height = generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 6).await; node_a.sync_wallets().unwrap(); node_b.sync_wallets().unwrap(); + wait_for_node_tip(&node_a, new_height).await; + wait_for_node_tip(&node_b, new_height).await; let expected_node_a_balance = expected_node_a_balance + amount_to_send_sats; let expected_node_b_balance_lower = expected_node_b_balance_lower - amount_to_send_sats; @@ -835,12 +838,12 @@ async fn onchain_send_receive() { let addr_b = node_b.onchain_payment().new_address().unwrap(); let txid = node_a.onchain_payment().send_all_to_address(&addr_b, true, None).unwrap(); - generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 6).await; + let new_height = generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 6).await; wait_for_tx(&electrsd.client, txid).await; - node_a.sync_wallets().unwrap(); node_b.sync_wallets().unwrap(); - + wait_for_node_tip(&node_a, new_height).await; + wait_for_node_tip(&node_b, new_height).await; let expected_node_b_balance_lower = expected_node_b_balance_lower + expected_node_a_balance; let expected_node_b_balance_upper = expected_node_b_balance_upper + expected_node_a_balance; let expected_node_a_balance = 0; @@ -858,11 +861,13 @@ async fn onchain_send_receive() { let addr_b = node_b.onchain_payment().new_address().unwrap(); let txid = node_a.onchain_payment().send_all_to_address(&addr_b, false, None).unwrap(); - generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 6).await; + let new_height = generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 6).await; wait_for_tx(&electrsd.client, txid).await; node_a.sync_wallets().unwrap(); node_b.sync_wallets().unwrap(); + wait_for_node_tip(&node_a, new_height).await; + wait_for_node_tip(&node_b, new_height).await; let expected_node_b_balance_lower = expected_node_b_balance_lower + reserve_amount_sat; let expected_node_b_balance_upper = expected_node_b_balance_upper + reserve_amount_sat; @@ -979,10 +984,11 @@ async fn onchain_send_all_retains_reserve() { let txid = node_a.onchain_payment().send_all_to_address(&addr_b, true, None).unwrap(); wait_for_tx(&electrsd.client, txid).await; - generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 6).await; - + let new_height = generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 6).await; node_a.sync_wallets().unwrap(); node_b.sync_wallets().unwrap(); + wait_for_node_tip(&node_a, new_height).await; + wait_for_node_tip(&node_b, new_height).await; // Check node a sent all and node b received it assert_eq!(node_a.list_balances().spendable_onchain_balance_sats, 0); assert!(((premine_amount_sat * 2 - onchain_fee_buffer_sat)..=(premine_amount_sat * 2)) @@ -997,16 +1003,20 @@ async fn onchain_send_all_retains_reserve() { .parse() .unwrap(); wait_for_tx(&electrsd.client, txid).await; - generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 6).await; + let new_height = generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 6).await; node_a.sync_wallets().unwrap(); node_b.sync_wallets().unwrap(); + wait_for_node_tip(&node_a, new_height).await; + wait_for_node_tip(&node_b, new_height).await; assert_eq!(node_a.list_balances().spendable_onchain_balance_sats, reserve_amount_sat); // Open a channel. open_channel(&node_b, &node_a, premine_amount_sat, false, &electrsd).await; - generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 6).await; + let new_height = generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 6).await; node_a.sync_wallets().unwrap(); node_b.sync_wallets().unwrap(); + wait_for_node_tip(&node_a, new_height).await; + wait_for_node_tip(&node_b, new_height).await; expect_channel_ready_event!(node_a, node_b.node_id()); expect_channel_ready_event!(node_b, node_a.node_id()); @@ -1020,10 +1030,12 @@ async fn onchain_send_all_retains_reserve() { let txid = node_b.onchain_payment().send_all_to_address(&addr_a, true, None).unwrap(); wait_for_tx(&electrsd.client, txid).await; - generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 6).await; + let new_height = generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 6).await; node_a.sync_wallets().unwrap(); node_b.sync_wallets().unwrap(); + wait_for_node_tip(&node_a, new_height).await; + wait_for_node_tip(&node_b, new_height).await; // Check node b sent all and node a received it assert_eq!(node_b.list_balances().total_onchain_balance_sats, reserve_amount_sat); @@ -1068,9 +1080,9 @@ async fn onchain_wallet_recovery() { .unwrap(); wait_for_tx(&electrsd.client, txid).await; - generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 1).await; - + let new_height = generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 1).await; original_node.sync_wallets().unwrap(); + wait_for_node_tip(&original_node, new_height).await; assert_eq!( original_node.list_balances().spendable_onchain_balance_sats, premine_amount_sat * 2 @@ -1106,9 +1118,9 @@ async fn onchain_wallet_recovery() { .unwrap(); wait_for_tx(&electrsd.client, txid).await; - generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 1).await; - + let new_height = generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 1).await; recovered_node.sync_wallets().unwrap(); + wait_for_node_tip(&recovered_node, new_height).await; assert_eq!( recovered_node.list_balances().spendable_onchain_balance_sats, premine_amount_sat * 3 @@ -1677,10 +1689,12 @@ async fn splice_channel() { open_channel(&node_a, &node_b, 4_000_000, false, &electrsd).await; // Open a channel with Node A contributing the funding - generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 6).await; + let new_height = generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 6).await; node_a.sync_wallets().unwrap(); node_b.sync_wallets().unwrap(); + wait_for_node_tip(&node_a, new_height).await; + wait_for_node_tip(&node_b, new_height).await; let user_channel_id_a = expect_channel_ready_event!(node_a, node_b.node_id()); let user_channel_id_b = expect_channel_ready_event!(node_b, node_a.node_id()); @@ -1777,7 +1791,9 @@ async fn splice_channel() { expect_payment_received_event!(node_a, amount_msat); // Mine a block to give time for the HTLC to resolve - generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 1).await; + let new_height = generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 1).await; + wait_for_node_tip(&node_a, new_height).await; + wait_for_node_tip(&node_b, new_height).await; assert_eq!( node_a.list_balances().total_lightning_balance_sats, @@ -2265,10 +2281,12 @@ async fn simple_bolt12_send_receive() { node_a.sync_wallets().unwrap(); open_channel(&node_a, &node_b, 4_000_000, true, &electrsd).await; - generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 6).await; + let new_height = generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 6).await; node_a.sync_wallets().unwrap(); node_b.sync_wallets().unwrap(); + wait_for_node_tip(&node_a, new_height).await; + wait_for_node_tip(&node_b, new_height).await; expect_channel_ready_event!(node_a, node_b.node_id()); expect_channel_ready_event!(node_b, node_a.node_id()); @@ -2550,12 +2568,16 @@ async fn async_payment() { ) .await; - generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 6).await; + let new_height = generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 6).await; node_sender.sync_wallets().unwrap(); node_sender_lsp.sync_wallets().unwrap(); node_receiver_lsp.sync_wallets().unwrap(); node_receiver.sync_wallets().unwrap(); + wait_for_node_tip(&node_sender, new_height).await; + wait_for_node_tip(&node_sender_lsp, new_height).await; + wait_for_node_tip(&node_receiver_lsp, new_height).await; + wait_for_node_tip(&node_receiver, new_height).await; expect_channel_ready_event!(node_sender, node_sender_lsp.node_id()); expect_channel_ready_events!( @@ -2752,10 +2774,12 @@ async fn generate_bip21_uri() { node_a.sync_wallets().unwrap(); open_channel(&node_a, &node_b, 4_000_000, true, &electrsd).await; - generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 6).await; + let new_height = generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 6).await; node_a.sync_wallets().unwrap(); node_b.sync_wallets().unwrap(); + wait_for_node_tip(&node_a, new_height).await; + wait_for_node_tip(&node_b, new_height).await; expect_channel_ready_event!(node_a, node_b.node_id()); expect_channel_ready_event!(node_b, node_a.node_id()); @@ -2804,10 +2828,12 @@ async fn unified_send_receive_bip21_uri() { node_a.sync_wallets().unwrap(); open_channel(&node_a, &node_b, 4_000_000, true, &electrsd).await; - generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 6).await; + let new_height = generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 6).await; node_a.sync_wallets().unwrap(); node_b.sync_wallets().unwrap(); + wait_for_node_tip(&node_a, new_height).await; + wait_for_node_tip(&node_b, new_height).await; expect_channel_ready_event!(node_a, node_b.node_id()); expect_channel_ready_event!(node_b, node_a.node_id()); @@ -2886,11 +2912,13 @@ async fn unified_send_receive_bip21_uri() { }, }; - generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 6).await; + let new_height = generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 6).await; wait_for_tx(&electrsd.client, txid).await; node_a.sync_wallets().unwrap(); node_b.sync_wallets().unwrap(); + wait_for_node_tip(&node_a, new_height).await; + wait_for_node_tip(&node_b, new_height).await; assert_eq!(node_b.list_balances().total_onchain_balance_sats, 800_000); assert_eq!(node_b.list_balances().total_lightning_balance_sats, 200_000); @@ -2970,9 +2998,11 @@ async fn do_lsps2_client_service_integration(client_trusts_lsp: bool) { println!("Opening channel payer_node -> service_node!"); open_channel(&payer_node, &service_node, 5_000_000, false, &electrsd).await; - generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 6).await; + let new_height = generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 6).await; service_node.sync_wallets().unwrap(); payer_node.sync_wallets().unwrap(); + wait_for_node_tip(&service_node, new_height).await; + wait_for_node_tip(&payer_node, new_height).await; expect_channel_ready_event!(payer_node, service_node.node_id()); expect_channel_ready_event!(service_node, payer_node.node_id()); @@ -3161,9 +3191,11 @@ async fn spontaneous_send_with_custom_preimage() { node_a.sync_wallets().unwrap(); node_b.sync_wallets().unwrap(); open_channel(&node_a, &node_b, 500_000, true, &electrsd).await; - generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 6).await; + let new_height = generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 6).await; node_a.sync_wallets().unwrap(); node_b.sync_wallets().unwrap(); + wait_for_node_tip(&node_a, new_height).await; + wait_for_node_tip(&node_b, new_height).await; expect_channel_ready_event!(node_a, node_b.node_id()); expect_channel_ready_event!(node_b, node_a.node_id()); @@ -3371,9 +3403,11 @@ async fn lsps2_client_trusts_lsp() { // Check the nodes pick up on the confirmed funding tx now. wait_for_tx(&electrsd.client, funding_txo.txid).await; - generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 6).await; + let new_height = generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 6).await; service_node.sync_wallets().unwrap(); client_node.sync_wallets().unwrap(); + wait_for_node_tip(&service_node, new_height).await; + wait_for_node_tip(&payer_node, new_height).await; assert_eq!( client_node .list_channels() @@ -3465,9 +3499,11 @@ async fn lsps2_lsp_trusts_client_but_client_does_not_claim() { // Open a channel payer -> service that will allow paying the JIT invoice open_channel(&payer_node, &service_node, 5_000_000, false, &electrsd).await; - generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 6).await; + let new_height = generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 6).await; service_node.sync_wallets().unwrap(); payer_node.sync_wallets().unwrap(); + wait_for_node_tip(&service_node, new_height).await; + wait_for_node_tip(&payer_node, new_height).await; expect_channel_ready_event!(payer_node, service_node.node_id()); expect_channel_ready_event!(service_node, payer_node.node_id()); @@ -3500,9 +3536,11 @@ async fn lsps2_lsp_trusts_client_but_client_does_not_claim() { // Check the nodes pick up on the confirmed funding tx now. wait_for_tx(&electrsd.client, funding_txo.txid).await; - generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 6).await; + let new_height = generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 6).await; service_node.sync_wallets().unwrap(); client_node.sync_wallets().unwrap(); + wait_for_node_tip(&service_node, new_height).await; + wait_for_node_tip(&client_node, new_height).await; assert_eq!( client_node .list_channels() @@ -3563,9 +3601,11 @@ async fn payment_persistence_after_restart() { // Open a large channel from node_a to node_b let channel_amount_sat = 5_000_000; open_channel(&node_a, &node_b, channel_amount_sat, true, &electrsd).await; - generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 6).await; + let new_height = generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 6).await; node_a.sync_wallets().unwrap(); node_b.sync_wallets().unwrap(); + wait_for_node_tip(&node_a, new_height).await; + wait_for_node_tip(&node_b, new_height).await; expect_channel_ready_event!(node_a, node_b.node_id()); expect_channel_ready_event!(node_b, node_a.node_id()); @@ -3922,9 +3962,11 @@ async fn onchain_fee_bump_rbf() { } // Confirm the transaction and try to bump again (should fail) - generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 6).await; + let new_height = generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 6).await; node_a.sync_wallets().unwrap(); node_b.sync_wallets().unwrap(); + wait_for_node_tip(&node_a, new_height).await; + wait_for_node_tip(&node_b, new_height).await; assert_eq!( Err(NodeError::InvalidPaymentId), @@ -4033,10 +4075,12 @@ async fn open_channel_with_all_with_anchors() { let funding_txo = open_channel_with_all(&node_a, &node_b, false, &electrsd).await; - generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 6).await; + let new_height = generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 6).await; node_a.sync_wallets().unwrap(); node_b.sync_wallets().unwrap(); + wait_for_node_tip(&node_a, new_height).await; + wait_for_node_tip(&node_b, new_height).await; let _user_channel_id_a = expect_channel_ready_event!(node_a, node_b.node_id()); let _user_channel_id_b = expect_channel_ready_event!(node_b, node_a.node_id()); @@ -4187,11 +4231,13 @@ async fn open_channel_variants_reserve_funds_for_anchor_peers() { opened_with_all_cases.push((variant, node_a, node_b, funding_txo_a)); } - generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 6).await; + let new_height = generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 6).await; for (variant, node_a, node_b, funding_txo) in opened_with_all_cases { node_a.sync_wallets().unwrap(); node_b.sync_wallets().unwrap(); + wait_for_node_tip(&node_a, new_height).await; + wait_for_node_tip(&node_b, new_height).await; let _user_channel_id_a = expect_channel_ready_event!(node_a, node_b.node_id()); let _user_channel_id_b = expect_channel_ready_event!(node_b, node_a.node_id()); @@ -4243,10 +4289,12 @@ async fn splice_in_with_all_balance() { // Open a channel with a fixed amount first let funding_txo = open_channel(&node_a, &node_b, channel_amount_sat, false, &electrsd).await; - generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 6).await; + let new_height = generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 6).await; node_a.sync_wallets().unwrap(); node_b.sync_wallets().unwrap(); + wait_for_node_tip(&node_a, new_height).await; + wait_for_node_tip(&node_b, new_height).await; let user_channel_id_a = expect_channel_ready_event!(node_a, node_b.node_id()); let _user_channel_id_b = expect_channel_ready_event!(node_b, node_a.node_id()); @@ -4262,10 +4310,12 @@ async fn splice_in_with_all_balance() { // Splice in with all remaining on-chain funds splice_in_with_all(&node_a, &node_b, &user_channel_id_a, &electrsd).await; - generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 6).await; + let new_height = generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 6).await; node_a.sync_wallets().unwrap(); node_b.sync_wallets().unwrap(); + wait_for_node_tip(&node_a, new_height).await; + wait_for_node_tip(&node_b, new_height).await; let _user_channel_id_a2 = expect_channel_ready_event!(node_a, node_b.node_id()); let _user_channel_id_b2 = expect_channel_ready_event!(node_b, node_a.node_id());