From 0cd84fb4e610dc756b934e040fc11ec135130f29 Mon Sep 17 00:00:00 2001 From: Alexander Shevtsov Date: Wed, 3 Jun 2026 01:06:18 +0200 Subject: [PATCH 1/5] Add CBF chain source stubs for starting Add stub methods/functions, add basic build and start of the CBF chain source as well as basic struct containing the fields which undoubtedtly are needed. --- Cargo.toml | 1 + src/chain/bitcoind.rs | 26 ++++ src/chain/cbf.rs | 306 ++++++++++++++++++++++++++++++++++++++++++ src/chain/mod.rs | 98 ++++++++++++-- src/lib.rs | 18 ++- src/wallet/mod.rs | 4 + 6 files changed, 438 insertions(+), 15 deletions(-) create mode 100644 src/chain/cbf.rs diff --git a/Cargo.toml b/Cargo.toml index b68a93f20..fd7c3a824 100755 --- a/Cargo.toml +++ b/Cargo.toml @@ -57,6 +57,7 @@ bdk_chain = { version = "0.23.0", default-features = false, features = ["std"] } bdk_esplora = { version = "0.22.0", default-features = false, features = ["async-https-rustls", "tokio"]} bdk_electrum = { version = "0.24.0", default-features = false, features = ["use-rustls-ring"]} bdk_wallet = { version = "2.3.0", default-features = false, features = ["std", "keys-bip39"]} +bip157 = { version = "0.6.0", 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/chain/bitcoind.rs b/src/chain/bitcoind.rs index 2582f32f6..d8d81ae8d 100644 --- a/src/chain/bitcoind.rs +++ b/src/chain/bitcoind.rs @@ -42,6 +42,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::types::{ChainMonitor, ChannelManager, DynStore, Sweeper, Wallet}; +use crate::BestBlock; use crate::{Error, NodeMetrics}; const CHAIN_POLLING_INTERVAL_SECS: u64 = 2; @@ -1341,6 +1342,7 @@ pub(crate) enum FeeRateEstimationMode { Conservative, } +#[derive(Clone)] pub(crate) struct ChainListener { pub(crate) onchain_wallet: Arc, pub(crate) channel_manager: Arc, @@ -1348,6 +1350,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..2ae654a0c --- /dev/null +++ b/src/chain/cbf.rs @@ -0,0 +1,306 @@ +use std::collections::{HashSet, VecDeque}; +use std::net::SocketAddr; +use std::sync::{Arc, Mutex}; +use std::time::Duration; + +use bip157::chain::ChainState; +use bip157::{ + Builder as CbfBuilder, Client, HashCheckpoint, Info, Node as CbfNode, Requester, TrustedPeer, + Warning, +}; +use bitcoin::{BlockHash, FeeRate, Script, ScriptBuf, Txid}; +use lightning::chain::WatchedOutput; +use tokio::sync::mpsc; + +use crate::chain::bitcoind::ChainListener; +use crate::chain::CbfFeeSourceConfig; +use crate::config::Config; +use crate::error::Error; +use crate::logger::{log_debug, log_error, log_info, LdkLogger, Logger}; +use crate::runtime::Runtime; +use crate::types::{ChainMonitor, ChannelManager, Sweeper, Wallet}; + +/// 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; + +const ESPLORA_TIMEOUT: u64 = 2; + +/// Runtime status of the underlying kyoto node. +enum CbfRuntimeStatus { + Started { requester: Requester }, + Stopped, +} + +/// Struct for holding cbf chain source +pub struct CbfChainSource { + /// Trusted peer addresses for kyoto's `Builder::add_peers`. + trusted_peers: Vec, + registered_scripts: Mutex>, + fee_source: FeeSource, + /// Tracks whether the kyoto node is running and holds the live requester. + cbf_runtime_status: Arc>, + /// Node configuration (network, storage path). + config: Arc, + logger: Arc, +} + +enum FeeSource { + /// Derive fee rates from the coinbase reward of recent blocks. Downloads full blocks in order to calculate fee estimation. + Cbf { block_fee_cache: Mutex> }, + /// 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, config: Arc, + logger: 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: Mutex::new(VecDeque::with_capacity(BLOCK_FEE_CACHE_CAPACITY)), + }, + }; + let registered_scripts = Mutex::new(HashSet::new()); + let cbf_runtime_status = Arc::new(Mutex::new(CbfRuntimeStatus::Stopped)); + Ok(Self { + trusted_peers, + fee_source, + registered_scripts, + cbf_runtime_status, + config, + logger, + }) + } + + fn build( + trusted_peers: &[TrustedPeer], config: &Config, logger: &Logger, + chain_listener: &ChainListener, + ) -> (CbfNode, Client) { + let mut cbf_builder = CbfBuilder::new(config.network); + + let data_dir = std::path::PathBuf::from(&config.storage_dir_path).join("bip157_data"); + cbf_builder = cbf_builder.data_dir(data_dir); + + if !trusted_peers.is_empty() { + cbf_builder = cbf_builder.add_peers(trusted_peers.to_vec()); + } + + cbf_builder = cbf_builder.required_peers(DEFAULT_REQUIRED_PEERS); + cbf_builder = cbf_builder.fetch_witness_data(); + cbf_builder = + cbf_builder.response_timeout(Duration::from_secs(DEFAULT_RESPONSE_TIMEOUT_SECS)); + + if let Some(header_cp) = Self::resume_checkpoint(logger, chain_listener) { + log_debug!( + logger, + "CBF builder: resuming from checkpoint height={}, hash={}", + header_cp.height, + header_cp.hash, + ); + cbf_builder = cbf_builder.chain_state(ChainState::Checkpoint(header_cp)); + } + + cbf_builder.build() + } + + 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())) + } + + pub(crate) fn start(&self, runtime: Arc, chain_listener: ChainListener) { + let (node, client) = + Self::build(&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 }; + } + + 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; + + 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 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), + )); + + match current_node.run().await { + Ok(()) => { + log_info!(restart_logger, "CBF node shut down cleanly."); + 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; + break; + } + log_error!( + restart_logger, + "CBF node exited with error (attempt {}/{}): {:?}. Restarting in {}ms.", + retries, + MAX_RESTART_RETRIES, + e, + backoff_ms, + ); + + tokio::time::sleep(Duration::from_millis(backoff_ms)).await; + backoff_ms = backoff_ms.saturating_mul(2); + + // Abort the old log consumers before rebuilding. + info_handle.abort(); + warn_handle.abort(); + + let (new_node, new_client) = Self::build( + &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_client; + + *restart_status.lock().expect("lock") = + CbfRuntimeStatus::Started { requester: new_requester }; + + current_node = new_node; + current_info_rx = new_info_rx; + current_warn_rx = new_warn_rx; + }, + } + } + }); + } + + pub(crate) fn stop(&self) { + todo!(); + } + + 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); + } + } + + pub(crate) fn process_kyoto_events( + &self, _stop_sync_receiver: tokio::sync::watch::Receiver<()>, _onchain_wallet: Arc, + _channel_manager: Arc, _chain_monitor: Arc, + _output_sweeper: Arc, + ) { + todo!(); + } + + pub(crate) fn register_tx(&self, _txid: &Txid, _script_pubkey: &Script) { + todo!(); + } + + pub(crate) fn register_output(&self, _output: WatchedOutput) { + todo!(); + // self.registered_scripts.lock().expect("lock").insert(output.script_pubkey.clone()); + } +} diff --git a/src/chain/mod.rs b/src/chain/mod.rs index cb8541be6..2b0c32681 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::{ @@ -82,6 +84,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>, @@ -93,6 +109,7 @@ enum ChainSourceKind { Esplora(EsploraChainSource), Electrum(ElectrumChainSource), Bitcoind(BitcoindChainSource), + Cbf(CbfChainSource), } impl ChainSource { @@ -184,11 +201,41 @@ 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, + 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, + Arc::clone(&config), + Arc::clone(&logger), + )?; + let kind = ChainSourceKind::Cbf(cbf_chain_source); + let registered_txids = Mutex::new(Vec::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(runtime, chain_listener); + }, _ => { // Nothing to do for other chain sources. }, @@ -223,6 +270,7 @@ impl ChainSource { ChainSourceKind::Esplora(_) => true, ChainSourceKind::Electrum { .. } => true, ChainSourceKind::Bitcoind { .. } => false, + ChainSourceKind::Cbf { .. } => false, } } @@ -249,9 +297,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; } }, @@ -272,9 +320,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; } }, @@ -289,6 +337,15 @@ impl ChainSource { ) .await }, + ChainSourceKind::Cbf(cbf_chain_source) => { + cbf_chain_source.process_kyoto_events( + stop_sync_receiver, + onchain_wallet, + channel_manager, + chain_monitor, + output_sweeper, + ); + }, } } @@ -331,7 +388,7 @@ impl ChainSource { log_trace!( logger, "Stopping background syncing on-chain wallet.", - ); + ); return; } _ = onchain_wallet_sync_interval.tick() => { @@ -345,7 +402,7 @@ impl ChainSource { Arc::clone(&channel_manager), Arc::clone(&chain_monitor), Arc::clone(&output_sweeper), - ).await; + ).await; } } } @@ -368,6 +425,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") + }, } } @@ -393,6 +453,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") + }, } } @@ -421,6 +484,9 @@ impl ChainSource { ) .await }, + ChainSourceKind::Cbf { .. } => { + todo!(); + }, } } @@ -435,6 +501,9 @@ impl ChainSource { ChainSourceKind::Bitcoind(bitcoind_chain_source) => { bitcoind_chain_source.update_fee_rate_estimates().await }, + ChainSourceKind::Cbf { .. } => { + todo!(); + }, } } @@ -463,6 +532,9 @@ impl ChainSource { ChainSourceKind::Bitcoind(bitcoind_chain_source) => { bitcoind_chain_source.process_broadcast_package(next_package).await }, + ChainSourceKind::Cbf { ..} => { + todo!(); + } } } } @@ -481,6 +553,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) { @@ -492,6 +567,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/lib.rs b/src/lib.rs index 2d1a0d958..24f49c92f 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -266,11 +266,19 @@ impl Node { self.config.network ); - // 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 + })?; // Block to ensure we update our fee rate cache once on startup let chain_source = Arc::clone(&self.chain_source); diff --git a/src/wallet/mod.rs b/src/wallet/mod.rs index 13b1f384f..d16447cfb 100644 --- a/src/wallet/mod.rs +++ b/src/wallet/mod.rs @@ -135,6 +135,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()); From fae9156efbd4106d68e680a6035cb0c05141ce12 Mon Sep 17 00:00:00 2001 From: Alexander Shevtsov Date: Wed, 3 Jun 2026 01:51:57 +0200 Subject: [PATCH 2/5] Add waiting for gossip propagation in tests Previously tests assumed that the chain source of the lightning node and is node which mines. This is not the case with CBF chain source which needs to wait until after mining a new block a new tips propagates to it. `wait_for_block` is made to return a new height and a new function `wait_for_node_tip` is added which waits until the given height is processed (returned via `status.best_block` ) on a given node. --- tests/common/mod.rs | 61 ++++++++++++---- tests/integration_tests_rust.rs | 120 ++++++++++++++++++++++---------- 2 files changed, 131 insertions(+), 50 deletions(-) diff --git a/tests/common/mod.rs b/tests/common/mod.rs index b701028c5..a5f5514f6 100644 --- a/tests/common/mod.rs +++ b/tests/common/mod.rs @@ -606,7 +606,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); @@ -615,9 +615,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(electrs, cur_height as usize + num).await; + let new_height = cur_height as usize + num; + wait_for_block(electrs, new_height).await; print!(" Done!"); println!("\n"); + return new_height; } pub(crate) fn invalidate_blocks(bitcoind: &BitcoindClient, num_blocks: usize) { @@ -688,6 +690,13 @@ pub(crate) async fn wait_for_outpoint_spend(electrs: &E, outpoin .await; } +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, @@ -982,7 +991,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(); @@ -1353,7 +1364,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(); @@ -1363,7 +1376,9 @@ 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; + 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(); @@ -1385,7 +1400,9 @@ 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; + 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(); @@ -1460,7 +1477,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(); @@ -1476,7 +1495,9 @@ pub(crate) async fn do_channel_full_cycle( assert_eq!(counterparty_node_id, node_a.node_id()); let cur_height = node_b.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_a, new_height).await; + wait_for_node_tip(&node_b, new_height).await; node_b.sync_wallets().unwrap(); node_a.sync_wallets().unwrap(); }, @@ -1489,7 +1510,9 @@ pub(crate) async fn do_channel_full_cycle( PendingSweepBalance::BroadcastAwaitingConfirmation { .. } => {}, _ => panic!("Unexpected balance state!"), } - 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_b.sync_wallets().unwrap(); node_a.sync_wallets().unwrap(); @@ -1499,7 +1522,9 @@ pub(crate) async fn do_channel_full_cycle( PendingSweepBalance::AwaitingThresholdConfirmations { .. } => {}, _ => 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_a, new_height).await; + wait_for_node_tip(&node_b, new_height).await; node_b.sync_wallets().unwrap(); node_a.sync_wallets().unwrap(); @@ -1517,7 +1542,9 @@ pub(crate) async fn do_channel_full_cycle( assert_eq!(counterparty_node_id, node_b.node_id()); let cur_height = node_a.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_a, new_height).await; + wait_for_node_tip(&node_b, new_height).await; node_a.sync_wallets().unwrap(); node_b.sync_wallets().unwrap(); }, @@ -1530,7 +1557,9 @@ pub(crate) async fn do_channel_full_cycle( PendingSweepBalance::BroadcastAwaitingConfirmation { .. } => {}, _ => panic!("Unexpected balance state!"), } - 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(); @@ -1540,7 +1569,9 @@ pub(crate) async fn do_channel_full_cycle( PendingSweepBalance::AwaitingThresholdConfirmations { .. } => {}, _ => 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_a, new_height).await; + wait_for_node_tip(&node_b, new_height).await; node_a.sync_wallets().unwrap(); node_b.sync_wallets().unwrap(); } else { @@ -1578,7 +1609,9 @@ 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 a3970b1c2..34db88337 100644 --- a/tests/integration_tests_rust.rs +++ b/tests/integration_tests_rust.rs @@ -24,7 +24,7 @@ use common::{ generate_listening_addresses, 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_tx, TestChainSource, TestStoreType, TestSyncStore, + wait_for_tx, wait_for_node_tip, TestChainSource, TestStoreType, TestSyncStore, }; use electrsd::corepc_node::Node as BitcoinD; use electrsd::ElectrsD; @@ -424,10 +424,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()); @@ -493,9 +494,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; @@ -531,12 +534,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; @@ -554,11 +557,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; @@ -607,10 +612,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)) @@ -625,16 +631,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()); @@ -648,10 +658,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); @@ -696,9 +708,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 @@ -734,9 +746,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 @@ -1014,10 +1026,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()); @@ -1059,7 +1073,7 @@ async fn splice_channel() { let txo = 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.client, &electrsd.client, 6).await; + generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 6).await; node_a.sync_wallets().unwrap(); node_b.sync_wallets().unwrap(); @@ -1099,7 +1113,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, @@ -1161,10 +1177,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()); @@ -1446,12 +1464,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!( @@ -1642,10 +1664,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()); @@ -1682,10 +1706,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()); @@ -1764,11 +1790,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); @@ -1848,7 +1876,7 @@ 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(); expect_channel_ready_event!(payer_node, service_node.node_id()); @@ -2039,9 +2067,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()); @@ -2249,9 +2279,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() @@ -2343,9 +2375,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()); @@ -2378,9 +2412,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() @@ -2441,9 +2477,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()); @@ -2753,9 +2791,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), @@ -2815,10 +2855,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()); @@ -2868,10 +2910,12 @@ async fn open_channel_with_all_without_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()); @@ -2921,10 +2965,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()); @@ -2940,10 +2986,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()); From ca6293e00dc96afcf418f7de2fd52bacd93b7cdc Mon Sep 17 00:00:00 2001 From: Alexander Shevtsov Date: Fri, 5 Jun 2026 04:24:09 +0200 Subject: [PATCH 3/5] Populate revealed spks for CBF Ask wallet for revealed spks, register them. Implement `Listen` trait ans add register_script method as well as implementation of registered scripts/outputs. --- src/builder.rs | 2 ++ src/chain/cbf.rs | 46 ++++++++++++++++++++++++++++++---------------- src/chain/mod.rs | 10 +++++++++- src/wallet/mod.rs | 26 ++++++++++++++++++++++++-- 4 files changed, 65 insertions(+), 19 deletions(-) diff --git a/src/builder.rs b/src/builder.rs index 5f616c4ce..f66ce15aa 100644 --- a/src/builder.rs +++ b/src/builder.rs @@ -1363,6 +1363,8 @@ fn build_with_store_internal( Arc::clone(&node_metrics), ) }, + //TODO add here an arm + // Some(ChainDataSoucrConfig::Cbf) Some(ChainDataSourceConfig::Bitcoind { rpc_host, rpc_port, diff --git a/src/chain/cbf.rs b/src/chain/cbf.rs index 2ae654a0c..72b6abf95 100644 --- a/src/chain/cbf.rs +++ b/src/chain/cbf.rs @@ -5,7 +5,7 @@ use std::time::Duration; use bip157::chain::ChainState; use bip157::{ - Builder as CbfBuilder, Client, HashCheckpoint, Info, Node as CbfNode, Requester, TrustedPeer, + Builder as KyotoBuilder, Client, HashCheckpoint, Info, Node as KyotoNode, Requester, TrustedPeer, Warning, }; use bitcoin::{BlockHash, FeeRate, Script, ScriptBuf, Txid}; @@ -109,23 +109,24 @@ impl CbfChainSource { }) } + //builds kyoto fn build( trusted_peers: &[TrustedPeer], config: &Config, logger: &Logger, chain_listener: &ChainListener, - ) -> (CbfNode, Client) { - let mut cbf_builder = CbfBuilder::new(config.network); + ) -> (KyotoNode, Client) { + let mut kyoto_builder = KyotoBuilder::new(config.network); let data_dir = std::path::PathBuf::from(&config.storage_dir_path).join("bip157_data"); - cbf_builder = cbf_builder.data_dir(data_dir); + kyoto_builder = kyoto_builder.data_dir(data_dir); if !trusted_peers.is_empty() { - cbf_builder = cbf_builder.add_peers(trusted_peers.to_vec()); + kyoto_builder = kyoto_builder.add_peers(trusted_peers.to_vec()); } - cbf_builder = cbf_builder.required_peers(DEFAULT_REQUIRED_PEERS); - cbf_builder = cbf_builder.fetch_witness_data(); - cbf_builder = - cbf_builder.response_timeout(Duration::from_secs(DEFAULT_RESPONSE_TIMEOUT_SECS)); + 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) = Self::resume_checkpoint(logger, chain_listener) { log_debug!( @@ -134,10 +135,11 @@ impl CbfChainSource { header_cp.height, header_cp.hash, ); - cbf_builder = cbf_builder.chain_state(ChainState::Checkpoint(header_cp)); + kyoto_builder = kyoto_builder.chain_state(ChainState::Checkpoint(header_cp)); } - cbf_builder.build() + + kyoto_builder.build() } fn resume_checkpoint( @@ -173,6 +175,12 @@ impl CbfChainSource { } pub(crate) fn start(&self, runtime: Arc, chain_listener: ChainListener) { + //we populate registered scripts with all the scripts from the onchain wallet + for script in chain_listener.onchain_wallet.list_revealed_scripts() { + self.register_script(script); + } + + let (node, client) = Self::build(&self.trusted_peers, &self.config, &self.logger, &chain_listener); let Client { requester, info_rx, warn_rx, event_rx: _ } = client; @@ -292,15 +300,21 @@ impl CbfChainSource { _channel_manager: Arc, _chain_monitor: Arc, _output_sweeper: Arc, ) { + //here we need to calculate chain update and feed to all listeners todo!(); } - pub(crate) fn register_tx(&self, _txid: &Txid, _script_pubkey: &Script) { - todo!(); + 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) { - todo!(); - // self.registered_scripts.lock().expect("lock").insert(output.script_pubkey.clone()); + pub(crate) fn register_output(&self, output: WatchedOutput) { + self.registered_scripts.lock().expect("lock").insert(output.script_pubkey); } + + pub(crate) fn register_script(&self, script: ScriptBuf) { + self.registered_scripts.lock().expect("lock").insert(script); + } + + } diff --git a/src/chain/mod.rs b/src/chain/mod.rs index 2b0c32681..7ba9d5d79 100644 --- a/src/chain/mod.rs +++ b/src/chain/mod.rs @@ -14,7 +14,7 @@ use std::collections::HashMap; use std::sync::{Arc, Mutex, RwLock}; use std::time::Duration; -use bitcoin::{Script, Txid}; +use bitcoin::{Script, ScriptBuf, Txid}; use lightning::chain::{BlockLocator, Filter}; use crate::chain::bitcoind::{BitcoindChainSource, ChainListener, UtxoSourceClient}; @@ -261,6 +261,14 @@ impl ChainSource { } } + pub(crate) fn register_script(&self, script: ScriptBuf) { + match &self.kind { + ChainSourceKind::Cbf(cbf) => cbf.register_script(script), + _ => {} // no-op: Esplora/Electrum/bitcoind don't need a watch set + } + } + + pub(crate) fn registered_txids(&self) -> Vec { self.registered_txids.lock().expect("lock").clone() } diff --git a/src/wallet/mod.rs b/src/wallet/mod.rs index d16447cfb..fce2e2755 100644 --- a/src/wallet/mod.rs +++ b/src/wallet/mod.rs @@ -35,7 +35,7 @@ use lightning::chain::chaininterface::{ BroadcasterInterface, INCREMENTAL_RELAY_FEE_SAT_PER_1000_WEIGHT, }; use lightning::chain::channelmonitor::ANTI_REORG_DELAY; -use lightning::chain::{BlockLocator, ClaimId, Listen}; +use lightning::chain::{BlockLocator, ClaimId, Listen, Filter}; use lightning::ln::channelmanager::PaymentId; use lightning::ln::inbound_payment::ExpandedKey; use lightning::ln::msgs::UnsignedGossipMessage; @@ -223,7 +223,26 @@ impl Wallet { Ok(()) } - fn update_payment_store<'a>( + pub(crate) fn list_revealed_scripts(&self) -> Vec { + self.inner + .lock() + .expect("lock") + .spk_index() + .revealed_spks(..) + .map(|((_keychain, _index), spk)| spk) + .collect() + } + + + /// Register scripts that BDK revealed at index time (e.g. change outputs, which `create_tx` + /// only peeks) with the chain source's watch set. No-op for non-CBF backends. + fn register_revealed_scripts(&self, _locked_wallet: &PersistedWallet) { + // TODO(cbf): diff `last_revealed_index(keychain)` against a per-keychain cursor and + // `chain_source.register_script(spk)` the delta for both keychains. + todo!() + } + + fn update_payment_store<'a>( &self, locked_wallet: &'a mut PersistedWallet, mut events: Vec, ) -> Result<(), Error> { @@ -488,6 +507,7 @@ impl Wallet { log_error!(self.logger, "Failed to persist wallet: {}", e); Error::PersistenceFailed })?; + self.chain_source.register_script(address_info.script_pubkey()); Ok(address_info.address) } @@ -500,6 +520,7 @@ impl Wallet { log_error!(self.logger, "Failed to persist wallet: {}", e); Error::PersistenceFailed })?; + self.chain_source.register_script(address_info.script_pubkey()); Ok(address_info.address) } @@ -1076,6 +1097,7 @@ impl Wallet { log_error!(self.logger, "Failed to persist wallet: {}", e); () })?; + self.chain_source.register_script(address_info.script_pubkey()); Ok(address_info.address.script_pubkey()) } From 567bd422761ede67c9709c65fa8e65d120fbce3b Mon Sep 17 00:00:00 2001 From: Alexander Shevtsov Date: Sat, 6 Jun 2026 22:12:37 +0200 Subject: [PATCH 4/5] Add sender and listener of `ChainOp`s MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit When `process_kyoto_events` processes events it decides whether we need to take any action (e.g. apply block). These actions are sent to a new abstraction — `BlockApplicator` which holds `ChainListener` (which in turn has wallets and can apply blocks / filtered blocks). This `BlockApplicator` has to have a receiver of a channel (and `process_kyoto_events` has to have a sender to this channel. Thus we cannot create them in `new`, because we would own them at `CbfChainSource`, so they are created in `start`. Also this commit ran `cargo fmt --all` which was missed previously. --- src/builder.rs | 4 +- src/chain/cbf.rs | 212 +++++++++++++++++++++++--------- src/chain/mod.rs | 28 ++--- src/wallet/mod.rs | 43 ++++--- tests/common/mod.rs | 69 ++++++----- tests/integration_tests_rust.rs | 112 ++++++++--------- 6 files changed, 281 insertions(+), 187 deletions(-) diff --git a/src/builder.rs b/src/builder.rs index f66ce15aa..a61f0e0e6 100644 --- a/src/builder.rs +++ b/src/builder.rs @@ -1363,8 +1363,8 @@ fn build_with_store_internal( Arc::clone(&node_metrics), ) }, - //TODO add here an arm - // Some(ChainDataSoucrConfig::Cbf) + //TODO add here an arm + // Some(ChainDataSoucrConfig::Cbf) Some(ChainDataSourceConfig::Bitcoind { rpc_host, rpc_port, diff --git a/src/chain/cbf.rs b/src/chain/cbf.rs index 72b6abf95..d9d70cc5c 100644 --- a/src/chain/cbf.rs +++ b/src/chain/cbf.rs @@ -5,12 +5,13 @@ use std::time::Duration; use bip157::chain::ChainState; use bip157::{ - Builder as KyotoBuilder, Client, HashCheckpoint, Info, Node as KyotoNode, Requester, TrustedPeer, - Warning, + chain::BlockHeaderChanges, error::FetchBlockError, Builder as KyotoBuilder, Client, Event, + HashCheckpoint, Header, IndexedBlock, Info, Node as KyotoNode, Requester, TrustedPeer, Warning, }; use bitcoin::{BlockHash, FeeRate, Script, ScriptBuf, Txid}; -use lightning::chain::WatchedOutput; -use tokio::sync::mpsc; +use lightning::chain::{Listen, WatchedOutput}; + +use tokio::sync::{mpsc, oneshot}; use crate::chain::bitcoind::ChainListener; use crate::chain::CbfFeeSourceConfig; @@ -50,7 +51,7 @@ enum CbfRuntimeStatus { pub struct CbfChainSource { /// Trusted peer addresses for kyoto's `Builder::add_peers`. trusted_peers: Vec, - registered_scripts: Mutex>, + registered_scripts: Arc>>, fee_source: FeeSource, /// Tracks whether the kyoto node is running and holds the live requester. cbf_runtime_status: Arc>, @@ -59,6 +60,36 @@ pub struct CbfChainSource { logger: Arc, } +enum ChainOp { + ConnectFull { block_rx: oneshot::Receiver> }, + ConnectFiltered { header: Header, height: u32 }, + //Reorg { /* accepted / reorganized from BlockHeaderChanges */ }, +} + +struct BlockApplicator { + chain_listener: ChainListener, + ops_rx: mpsc::UnboundedReceiver, + logger: Arc, +} + +impl BlockApplicator { + async fn run(mut self) { + while let Some(op) = self.ops_rx.recv().await { + match op { + ChainOp::ConnectFull { block_rx } => match block_rx.await { + Ok(Ok(ib)) => self.chain_listener.block_connected(&ib.block, ib.height), + Ok(Err(e)) => log_error!(self.logger, "block fetch failed: {:?}", e), + Err(_) => log_error!(self.logger, "block oneshot dropped"), + }, + ChainOp::ConnectFiltered { header, height } => { + self.chain_listener.filtered_block_connected(&header, &[], height) + }, + //ChainOp::Reorg { .. } => {}, + } + } + } +} + enum FeeSource { /// Derive fee rates from the coinbase reward of recent blocks. Downloads full blocks in order to calculate fee estimation. Cbf { block_fee_cache: Mutex> }, @@ -70,6 +101,17 @@ enum FeeSource { Electrum { server_url: String }, } +impl FeeSource { + fn insert_cached_block(&self, block_hash: BlockHash, fee_rate: FeeRate) { + match &self { + Self::Cbf { block_fee_cache } => { + block_fee_cache.lock().expect("lock").push_back((block_hash, fee_rate)); + }, + _ => {}, + } + } +} + impl CbfChainSource { pub(crate) fn new( peers: Vec, fee_source_config: Option, config: Arc, @@ -97,7 +139,7 @@ impl CbfChainSource { block_fee_cache: Mutex::new(VecDeque::with_capacity(BLOCK_FEE_CACHE_CAPACITY)), }, }; - let registered_scripts = Mutex::new(HashSet::new()); + let registered_scripts = Arc::new(Mutex::new(HashSet::new())); let cbf_runtime_status = Arc::new(Mutex::new(CbfRuntimeStatus::Stopped)); Ok(Self { trusted_peers, @@ -109,8 +151,7 @@ impl CbfChainSource { }) } - //builds kyoto - fn build( + fn build_kyoto( trusted_peers: &[TrustedPeer], config: &Config, logger: &Logger, chain_listener: &ChainListener, ) -> (KyotoNode, Client) { @@ -128,7 +169,7 @@ impl CbfChainSource { kyoto_builder = kyoto_builder.response_timeout(Duration::from_secs(DEFAULT_RESPONSE_TIMEOUT_SECS)); - if let Some(header_cp) = Self::resume_checkpoint(logger, chain_listener) { + if let Some(header_cp) = resume_checkpoint(logger, chain_listener) { log_debug!( logger, "CBF builder: resuming from checkpoint height={}, hash={}", @@ -138,52 +179,18 @@ impl CbfChainSource { kyoto_builder = kyoto_builder.chain_state(ChainState::Checkpoint(header_cp)); } - kyoto_builder.build() } - 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())) - } - pub(crate) fn start(&self, runtime: Arc, chain_listener: ChainListener) { - //we populate registered scripts with all the scripts from the onchain wallet - for script in chain_listener.onchain_wallet.list_revealed_scripts() { - self.register_script(script); - } - + //populate registered scripts with all the scripts from the onchain wallet + for script in chain_listener.onchain_wallet.list_revealed_scripts() { + self.register_script(script); + } let (node, client) = - Self::build(&self.trusted_peers, &self.config, &self.logger, &chain_listener); - let Client { requester, info_rx, warn_rx, event_rx: _ } = 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"); @@ -194,6 +201,14 @@ impl CbfChainSource { *status = CbfRuntimeStatus::Started { requester }; } + let (ops_tx, ops_rx) = mpsc::unbounded_channel(); + let block_applicator = BlockApplicator { + chain_listener: chain_listener.clone(), + ops_rx, + logger: Arc::clone(&self.logger), + }; + runtime.spawn_background_task(block_applicator.run()); + log_info!(self.logger, "CBF chain source started."); let restart_status = Arc::clone(&self.cbf_runtime_status); @@ -201,11 +216,15 @@ impl CbfChainSource { 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_block_applicator = 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; @@ -219,6 +238,13 @@ impl CbfChainSource { Arc::clone(&restart_logger), )); + let event_handle = tokio::spawn(Self::process_kyoto_events( + current_event_rx, + Arc::clone(&restart_registered_scripts), + Arc::clone(&restart_cbf_runtime_status), + ops_tx.clone(), + )); + match current_node.run().await { Ok(()) => { log_info!(restart_logger, "CBF node shut down cleanly."); @@ -251,8 +277,9 @@ impl CbfChainSource { // Abort the old log consumers before rebuilding. info_handle.abort(); warn_handle.abort(); + event_handle.abort(); - let (new_node, new_client) = Self::build( + let (new_node, new_client) = Self::build_kyoto( &restart_peers, &restart_config, &restart_logger, @@ -262,7 +289,7 @@ impl CbfChainSource { requester: new_requester, info_rx: new_info_rx, warn_rx: new_warn_rx, - event_rx: _, + event_rx: new_event_rx, } = new_client; *restart_status.lock().expect("lock") = @@ -271,6 +298,7 @@ impl CbfChainSource { current_node = new_node; current_info_rx = new_info_rx; current_warn_rx = new_warn_rx; + current_event_rx = new_event_rx; }, } } @@ -295,13 +323,49 @@ impl CbfChainSource { } } - pub(crate) fn process_kyoto_events( - &self, _stop_sync_receiver: tokio::sync::watch::Receiver<()>, _onchain_wallet: Arc, - _channel_manager: Arc, _chain_monitor: Arc, - _output_sweeper: Arc, + async fn process_kyoto_events( + mut event_rx: mpsc::UnboundedReceiver, + registered_scripts: Arc>>, + cbf_runtime_status: Arc>, ops_tx: mpsc::UnboundedSender, ) { - //here we need to calculate chain update and feed to all listeners - todo!(); + while let Some(event) = event_rx.recv().await { + match event { + // match download + Event::IndexedFilter(indexed_filter) => { + let matched = indexed_filter + .contains_any(registered_scripts.lock().expect("lock").iter()); + if matched { + let rtm = &*cbf_runtime_status.lock().expect("lock"); + let requestor = match rtm { + CbfRuntimeStatus::Started { requester } => requester.clone(), + CbfRuntimeStatus::Stopped => { + //panic + // todo!(); + continue; + }, + }; + let block_rx = requestor + .request_block(indexed_filter.block_hash()) + .expect("cannot request block"); + let chop = ChainOp::ConnectFull { block_rx }; + //here we feed evets to the driver + ops_tx.send(chop); + } + }, + Event::FiltersSynced(sync_update) => { + todo!(); + }, + Event::ChainUpdate(BlockHeaderChanges::Connected(connected_blocks)) => { + todo!(); + }, + Event::ChainUpdate(BlockHeaderChanges::Reorganized { reorganized, accepted }) => { + todo!(); + }, + Event::ChainUpdate(BlockHeaderChanges::ForkAdded(fork)) => { + todo!(); + }, + } + } } pub(crate) fn register_tx(&self, _txid: &Txid, script_pubkey: &Script) { @@ -312,9 +376,37 @@ impl CbfChainSource { self.registered_scripts.lock().expect("lock").insert(output.script_pubkey); } - pub(crate) fn register_script(&self, script: ScriptBuf) { - self.registered_scripts.lock().expect("lock").insert(script); - } + pub(crate) fn register_script(&self, script: ScriptBuf) { + self.registered_scripts.lock().expect("lock").insert(script); + } +} + +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/mod.rs b/src/chain/mod.rs index 7ba9d5d79..ed5630206 100644 --- a/src/chain/mod.rs +++ b/src/chain/mod.rs @@ -261,13 +261,12 @@ impl ChainSource { } } - pub(crate) fn register_script(&self, script: ScriptBuf) { - match &self.kind { - ChainSourceKind::Cbf(cbf) => cbf.register_script(script), - _ => {} // no-op: Esplora/Electrum/bitcoind don't need a watch set - } - } - + pub(crate) fn register_script(&self, script: ScriptBuf) { + match &self.kind { + ChainSourceKind::Cbf(cbf) => cbf.register_script(script), + _ => {}, // no-op: Esplora/Electrum/bitcoind don't need a watch set + } + } pub(crate) fn registered_txids(&self) -> Vec { self.registered_txids.lock().expect("lock").clone() @@ -346,13 +345,14 @@ impl ChainSource { .await }, ChainSourceKind::Cbf(cbf_chain_source) => { - cbf_chain_source.process_kyoto_events( - stop_sync_receiver, - onchain_wallet, - channel_manager, - chain_monitor, - output_sweeper, - ); + todo!(); + // cbf_chain_source.process_kyoto_events( + // stop_sync_receiver, + // onchain_wallet, + // channel_manager, + // chain_monitor, + // output_sweeper, + // ); }, } } diff --git a/src/wallet/mod.rs b/src/wallet/mod.rs index fce2e2755..854bb312d 100644 --- a/src/wallet/mod.rs +++ b/src/wallet/mod.rs @@ -35,7 +35,7 @@ use lightning::chain::chaininterface::{ BroadcasterInterface, INCREMENTAL_RELAY_FEE_SAT_PER_1000_WEIGHT, }; use lightning::chain::channelmonitor::ANTI_REORG_DELAY; -use lightning::chain::{BlockLocator, ClaimId, Listen, Filter}; +use lightning::chain::{BlockLocator, ClaimId, Filter, Listen}; use lightning::ln::channelmanager::PaymentId; use lightning::ln::inbound_payment::ExpandedKey; use lightning::ln::msgs::UnsignedGossipMessage; @@ -223,26 +223,25 @@ impl Wallet { Ok(()) } - pub(crate) fn list_revealed_scripts(&self) -> Vec { - self.inner - .lock() - .expect("lock") - .spk_index() - .revealed_spks(..) - .map(|((_keychain, _index), spk)| spk) - .collect() - } - + pub(crate) fn list_revealed_scripts(&self) -> Vec { + self.inner + .lock() + .expect("lock") + .spk_index() + .revealed_spks(..) + .map(|((_keychain, _index), spk)| spk) + .collect() + } - /// Register scripts that BDK revealed at index time (e.g. change outputs, which `create_tx` - /// only peeks) with the chain source's watch set. No-op for non-CBF backends. - fn register_revealed_scripts(&self, _locked_wallet: &PersistedWallet) { - // TODO(cbf): diff `last_revealed_index(keychain)` against a per-keychain cursor and - // `chain_source.register_script(spk)` the delta for both keychains. - todo!() - } + /// Register scripts that BDK revealed at index time (e.g. change outputs, which `create_tx` + /// only peeks) with the chain source's watch set. No-op for non-CBF backends. + fn register_revealed_scripts(&self, _locked_wallet: &PersistedWallet) { + // TODO(cbf): diff `last_revealed_index(keychain)` against a per-keychain cursor and + // `chain_source.register_script(spk)` the delta for both keychains. + todo!() + } - fn update_payment_store<'a>( + fn update_payment_store<'a>( &self, locked_wallet: &'a mut PersistedWallet, mut events: Vec, ) -> Result<(), Error> { @@ -507,7 +506,7 @@ impl Wallet { log_error!(self.logger, "Failed to persist wallet: {}", e); Error::PersistenceFailed })?; - self.chain_source.register_script(address_info.script_pubkey()); + self.chain_source.register_script(address_info.script_pubkey()); Ok(address_info.address) } @@ -520,7 +519,7 @@ impl Wallet { log_error!(self.logger, "Failed to persist wallet: {}", e); Error::PersistenceFailed })?; - self.chain_source.register_script(address_info.script_pubkey()); + self.chain_source.register_script(address_info.script_pubkey()); Ok(address_info.address) } @@ -1097,7 +1096,7 @@ impl Wallet { log_error!(self.logger, "Failed to persist wallet: {}", e); () })?; - self.chain_source.register_script(address_info.script_pubkey()); + self.chain_source.register_script(address_info.script_pubkey()); Ok(address_info.address.script_pubkey()) } diff --git a/tests/common/mod.rs b/tests/common/mod.rs index a5f5514f6..78f0ac6cc 100644 --- a/tests/common/mod.rs +++ b/tests/common/mod.rs @@ -615,11 +615,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); - let new_height = cur_height as usize + num; + let new_height = cur_height as usize + num; wait_for_block(electrs, new_height).await; print!(" Done!"); println!("\n"); - return new_height; + return new_height; } pub(crate) fn invalidate_blocks(bitcoind: &BitcoindClient, num_blocks: usize) { @@ -691,10 +691,10 @@ pub(crate) async fn wait_for_outpoint_spend(electrs: &E, outpoin } 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; + 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 @@ -992,8 +992,8 @@ pub(crate) async fn do_channel_full_cycle( if !allow_0conf { 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; + wait_for_node_tip(&node_a, new_height).await; + wait_for_node_tip(&node_b, new_height).await; } node_a.sync_wallets().unwrap(); @@ -1365,8 +1365,8 @@ pub(crate) async fn do_channel_full_cycle( // Mine a block to give time for the HTLC to resolve 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; + 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(); @@ -1377,8 +1377,8 @@ pub(crate) async fn do_channel_full_cycle( expect_splice_negotiated_event!(node_b, node_a.node_id()); 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; + 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(); @@ -1401,8 +1401,8 @@ pub(crate) async fn do_channel_full_cycle( expect_splice_negotiated_event!(node_b, node_a.node_id()); 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; + 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(); @@ -1478,8 +1478,8 @@ pub(crate) async fn do_channel_full_cycle( wait_for_outpoint_spend(electrsd, funding_txo_b).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; + 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(); @@ -1495,9 +1495,10 @@ pub(crate) async fn do_channel_full_cycle( assert_eq!(counterparty_node_id, node_a.node_id()); let cur_height = node_b.status().current_best_block.height; let blocks_to_go = confirmation_height - cur_height; - let new_height = generate_blocks_and_wait(&bitcoind, electrsd, blocks_to_go as usize).await; - wait_for_node_tip(&node_a, new_height).await; - wait_for_node_tip(&node_b, new_height).await; + let new_height = + generate_blocks_and_wait(&bitcoind, electrsd, 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_b.sync_wallets().unwrap(); node_a.sync_wallets().unwrap(); }, @@ -1511,8 +1512,8 @@ pub(crate) async fn do_channel_full_cycle( _ => panic!("Unexpected balance state!"), } 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; + wait_for_node_tip(&node_a, new_height).await; + wait_for_node_tip(&node_b, new_height).await; node_b.sync_wallets().unwrap(); node_a.sync_wallets().unwrap(); @@ -1523,8 +1524,8 @@ pub(crate) async fn do_channel_full_cycle( _ => panic!("Unexpected balance state!"), } let new_height = generate_blocks_and_wait(&bitcoind, electrsd, 5).await; - wait_for_node_tip(&node_a, new_height).await; - wait_for_node_tip(&node_b, new_height).await; + wait_for_node_tip(&node_a, new_height).await; + wait_for_node_tip(&node_b, new_height).await; node_b.sync_wallets().unwrap(); node_a.sync_wallets().unwrap(); @@ -1542,9 +1543,10 @@ pub(crate) async fn do_channel_full_cycle( assert_eq!(counterparty_node_id, node_b.node_id()); let cur_height = node_a.status().current_best_block.height; let blocks_to_go = confirmation_height - cur_height; - let new_height = generate_blocks_and_wait(&bitcoind, electrsd, blocks_to_go as usize).await; - wait_for_node_tip(&node_a, new_height).await; - wait_for_node_tip(&node_b, new_height).await; + let new_height = + generate_blocks_and_wait(&bitcoind, electrsd, 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(); }, @@ -1558,8 +1560,8 @@ pub(crate) async fn do_channel_full_cycle( _ => panic!("Unexpected balance state!"), } 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; + 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(); @@ -1570,8 +1572,8 @@ pub(crate) async fn do_channel_full_cycle( _ => panic!("Unexpected balance state!"), } let new_height = generate_blocks_and_wait(&bitcoind, electrsd, 5).await; - wait_for_node_tip(&node_a, new_height).await; - wait_for_node_tip(&node_b, new_height).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(); } else { @@ -1609,9 +1611,10 @@ pub(crate) async fn do_channel_full_cycle( assert_eq!(node_a_blocks_to_go, node_b_blocks_to_go); - 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; + 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 34db88337..0dba59cd1 100644 --- a/tests/integration_tests_rust.rs +++ b/tests/integration_tests_rust.rs @@ -24,7 +24,7 @@ use common::{ generate_listening_addresses, 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_tx, wait_for_node_tip, TestChainSource, TestStoreType, TestSyncStore, + wait_for_node_tip, wait_for_tx, TestChainSource, TestStoreType, TestSyncStore, }; use electrsd::corepc_node::Node as BitcoinD; use electrsd::ElectrsD; @@ -427,8 +427,8 @@ async fn onchain_send_receive() { 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; + 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()); @@ -497,8 +497,8 @@ async fn onchain_send_receive() { 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; + 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; @@ -538,8 +538,8 @@ async fn onchain_send_receive() { 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; + 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; @@ -562,8 +562,8 @@ async fn onchain_send_receive() { 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; + 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; @@ -615,8 +615,8 @@ async fn onchain_send_all_retains_reserve() { 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; + 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)) @@ -634,8 +634,8 @@ async fn onchain_send_all_retains_reserve() { 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; + 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. @@ -643,8 +643,8 @@ async fn onchain_send_all_retains_reserve() { 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; + 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()); @@ -662,8 +662,8 @@ async fn onchain_send_all_retains_reserve() { 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; + 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); @@ -710,7 +710,7 @@ async fn onchain_wallet_recovery() { 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; + wait_for_node_tip(&original_node, new_height).await; assert_eq!( original_node.list_balances().spendable_onchain_balance_sats, premine_amount_sat * 2 @@ -748,7 +748,7 @@ async fn onchain_wallet_recovery() { 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; + wait_for_node_tip(&recovered_node, new_height).await; assert_eq!( recovered_node.list_balances().spendable_onchain_balance_sats, premine_amount_sat * 3 @@ -1030,8 +1030,8 @@ async fn splice_channel() { 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; + 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()); @@ -1073,7 +1073,7 @@ async fn splice_channel() { let txo = 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.client, &electrsd.client, 6).await; + generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 6).await; node_a.sync_wallets().unwrap(); node_b.sync_wallets().unwrap(); @@ -1114,8 +1114,8 @@ async fn splice_channel() { // Mine a block to give time for the HTLC to resolve 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; + 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, @@ -1181,8 +1181,8 @@ async fn simple_bolt12_send_receive() { 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; + 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()); @@ -1470,10 +1470,10 @@ async fn async_payment() { 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; + 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!( @@ -1668,8 +1668,8 @@ async fn generate_bip21_uri() { 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; + 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()); @@ -1710,8 +1710,8 @@ async fn unified_send_receive_bip21_uri() { 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; + 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()); @@ -1795,8 +1795,8 @@ async fn unified_send_receive_bip21_uri() { 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; + 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); @@ -2070,8 +2070,8 @@ async fn spontaneous_send_with_custom_preimage() { 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; + 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()); @@ -2282,8 +2282,8 @@ async fn lsps2_client_trusts_lsp() { 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; + wait_for_node_tip(&service_node, new_height).await; + wait_for_node_tip(&payer_node, new_height).await; assert_eq!( client_node .list_channels() @@ -2378,8 +2378,8 @@ async fn lsps2_lsp_trusts_client_but_client_does_not_claim() { 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; + 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()); @@ -2415,8 +2415,8 @@ async fn lsps2_lsp_trusts_client_but_client_does_not_claim() { 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; + wait_for_node_tip(&service_node, new_height).await; + wait_for_node_tip(&client_node, new_height).await; assert_eq!( client_node .list_channels() @@ -2480,8 +2480,8 @@ async fn payment_persistence_after_restart() { 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; + 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()); @@ -2794,8 +2794,8 @@ async fn onchain_fee_bump_rbf() { 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; + wait_for_node_tip(&node_a, new_height).await; + wait_for_node_tip(&node_b, new_height).await; assert_eq!( Err(NodeError::InvalidPaymentId), @@ -2859,8 +2859,8 @@ async fn open_channel_with_all_with_anchors() { 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; + 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()); @@ -2914,8 +2914,8 @@ async fn open_channel_with_all_without_anchors() { 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; + 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()); @@ -2969,8 +2969,8 @@ async fn splice_in_with_all_balance() { 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; + 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()); @@ -2990,8 +2990,8 @@ async fn splice_in_with_all_balance() { 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; + 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()); From ce52bb982455c43c1b8aa51a2b1d79d2965ae2c4 Mon Sep 17 00:00:00 2001 From: Alexander Shevtsov Date: Tue, 9 Jun 2026 04:09:11 +0200 Subject: [PATCH 5/5] Implement `process_kyoto_events` and `ChainOp` Added 4 variants of `ChainOp`: - ConnectFull, - ConnectFiltered, - Disconnect, - Synced Now `process_kyoto_events` reacts to an event from kyoto and sends a `ChainOp` to listener (`BlockApplicator`). Note that on a filter with no match we still need to apply header and we need double check that header in the canonical chain is the relevant one and has not been reorged. Right now it is left as a todo, because of an upstream PR. Co-authored-by: febyeji --- src/chain/cbf.rs | 128 ++++++++++++++++++++++++++++++++++++----------- 1 file changed, 100 insertions(+), 28 deletions(-) diff --git a/src/chain/cbf.rs b/src/chain/cbf.rs index baca0ced5..b76650cf5 100644 --- a/src/chain/cbf.rs +++ b/src/chain/cbf.rs @@ -9,7 +9,7 @@ use bip157::{ HashCheckpoint, Header, IndexedBlock, Info, Node as KyotoNode, Requester, TrustedPeer, Warning, }; use bitcoin::{BlockHash, FeeRate, Script, ScriptBuf, Txid}; -use lightning::chain::{Listen, WatchedOutput}; +use lightning::chain::{BlockLocator, Listen, WatchedOutput}; use tokio::sync::{mpsc, oneshot}; @@ -61,9 +61,20 @@ pub struct CbfChainSource { } enum ChainOp { - ConnectFull { block_rx: oneshot::Receiver> }, - ConnectFiltered { header: Header, height: u32 }, - //Reorg { /* accepted / reorganized from BlockHeaderChanges */ }, + ConnectFull { + block_rx: oneshot::Receiver>, + }, + ConnectFiltered { + header: Header, + height: u32, + }, + Disconnect { + fork_point: BlockLocator, + }, + /// Marks reaching the chain tip. + Synced { + tip_height: u32, + }, } struct BlockApplicator { @@ -82,9 +93,16 @@ impl BlockApplicator { Err(_) => log_error!(self.logger, "block oneshot dropped"), }, ChainOp::ConnectFiltered { header, height } => { - self.chain_listener.filtered_block_connected(&header, &[], height) + self.chain_listener.filtered_block_connected(&header, &[], height); + }, + ChainOp::Disconnect { fork_point } => { + self.chain_listener.blocks_disconnected(fork_point); + }, + ChainOp::Synced { tip_height } => { + log_info!(self.logger, "CBF caught up to tip {}", tip_height); + // TODO: notify sync-completion waiters (start()/sync_wallets()/tests) once + // a notification primitive is plumbed through. }, - //ChainOp::Reorg { .. } => {}, } } } @@ -239,6 +257,7 @@ impl CbfChainSource { )); 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), @@ -351,45 +370,98 @@ impl CbfChainSource { } async fn process_kyoto_events( - mut event_rx: mpsc::UnboundedReceiver, + logger: Arc, mut event_rx: mpsc::UnboundedReceiver, registered_scripts: Arc>>, cbf_runtime_status: Arc>, ops_tx: mpsc::UnboundedSender, ) { while let Some(event) = event_rx.recv().await { match event { - // match download Event::IndexedFilter(indexed_filter) => { + let requester = match &*cbf_runtime_status.lock().expect("lock") { + CbfRuntimeStatus::Started { requester } => requester.clone(), + CbfRuntimeStatus::Stopped => { + //TODO should we panic here? what do we do if we have no requester? + continue; + }, + }; + let block_hash = indexed_filter.block_hash(); let matched = indexed_filter .contains_any(registered_scripts.lock().expect("lock").iter()); - if matched { - let rtm = &*cbf_runtime_status.lock().expect("lock"); - let requestor = match rtm { - CbfRuntimeStatus::Started { requester } => requester.clone(), - CbfRuntimeStatus::Stopped => { - //panic - // todo!(); + + let chop: ChainOp = if matched { + let block_rx = + requester.request_block(block_hash).expect("cannot request block"); + ChainOp::ConnectFull { block_rx } + } 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. right now this would block + //the further sync, as we cannot apply blocks in order. + //Future solution would use something like `get_header_by_hash`. + 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,); continue; }, - }; - let block_rx = requestor - .request_block(indexed_filter.block_hash()) - .expect("cannot request block"); - let chop = ChainOp::ConnectFull { block_rx }; - //here we feed evets to the driver - ops_tx.send(chop); + Err(e) => { + log_error!( + logger, + "Failed to fetch header at height {}: {:?}", + height, + e, + ); + continue; + }, + } + }; + if let Err(e) = ops_tx.send(chop) { + log_debug!(logger, "ops_rx gone: {}", e); } }, Event::FiltersSynced(sync_update) => { - todo!(); + //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(connected_blocks)) => { - todo!(); + Event::ChainUpdate(BlockHeaderChanges::Connected(indexed_header)) => { + log_debug!( + logger, + "Kyoto connected header at height {}", + indexed_header.height + ); }, - Event::ChainUpdate(BlockHeaderChanges::Reorganized { reorganized, accepted }) => { - todo!(); + 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)) => { - todo!(); + log_debug!(logger, "Kyoto added fork header at height {}", fork.height); }, } }