Add CBF documentation, on-chain reorg handling, and CI coverage

Author: febyeji

.github/workflows/rust.yml

@@ -72,11 +72,17 @@ jobs:
       - name: Build with UniFFI support on Rust ${{ matrix.toolchain }}
         if: matrix.build-uniffi
         run: cargo build --features uniffi --verbose --color always
+      - name: Build with CBF and UniFFI support on Rust ${{ matrix.toolchain }}
+        if: matrix.build-uniffi
+        run: cargo build --features "cbf uniffi" --verbose --color always
       - name: Check release build on Rust ${{ matrix.toolchain }}
         run: cargo check --release --verbose --color always
       - name: Check release build with UniFFI support on Rust ${{ matrix.toolchain }}
         if: matrix.build-uniffi
         run: cargo check --release --features uniffi --verbose --color always
+      - name: Check release build with CBF and UniFFI support on Rust ${{ matrix.toolchain }}
+        if: matrix.build-uniffi
+        run: cargo check --release --features "cbf uniffi" --verbose --color always
       - name: Test on Rust ${{ matrix.toolchain }}
         if: "matrix.platform != 'windows-latest'"
         run: |
@@ -85,6 +91,14 @@ jobs:
         if: "matrix.platform != 'windows-latest' && matrix.build-uniffi"
         run: |
           RUSTFLAGS="--cfg no_download --cfg cycle_tests" cargo test --features uniffi
+      - name: Test with CBF and UniFFI support on Rust ${{ matrix.toolchain }}
+        if: "matrix.platform != 'windows-latest' && matrix.build-uniffi"
+        run: |
+          RUSTFLAGS="--cfg no_download --cfg cycle_tests" cargo test --features "cbf uniffi" --lib arced_builder_can_set_cbf_chain_source
+      - name: Test with CBF support on Rust ${{ matrix.toolchain }}
+        if: "matrix.platform != 'windows-latest'"
+        run: |
+          RUSTFLAGS="--cfg no_download --cfg cycle_tests" cargo test --features cbf
 
   doc:
     name: Documentation

README.md

@@ -61,7 +61,7 @@ fn main() {
 LDK Node currently comes with a decidedly opinionated set of design choices:
 
 - On-chain data is handled by the integrated [BDK][bdk] wallet.
-- Chain data may currently be sourced from the Bitcoin Core RPC interface, or from an [Electrum][electrum] or [Esplora][esplora] server.
+- Chain data may currently be sourced from the Bitcoin Core RPC interface, from an [Electrum][electrum] or [Esplora][esplora] server, or via [compact block filters (BIP 157)][bip157] when the `cbf` feature is enabled.
 - Wallet and channel state may be persisted to an [SQLite][sqlite] database, to file system, or to a custom back-end to be implemented by the user.
 - Gossip data may be sourced via Lightning's peer-to-peer network or the [Rapid Gossip Sync](https://docs.rs/lightning-rapid-gossip-sync/*/lightning_rapid_gossip_sync/) protocol.
 - Entropy for the Lightning and on-chain wallets may be sourced from raw bytes or a [BIP39](https://github.com/bitcoin/bips/blob/master/bip-0039.mediawiki) mnemonic. In addition, LDK Node offers the means to generate and persist the entropy bytes to disk.
@@ -80,6 +80,7 @@ The Minimum Supported Rust Version (MSRV) is currently 1.85.0.
 [bdk]: https://bitcoindevkit.org/
 [electrum]: https://github.com/spesmilo/electrum-protocol
 [esplora]: https://github.com/Blockstream/esplora
+[bip157]: https://github.com/bitcoin/bips/blob/master/bip-0157.mediawiki
 [sqlite]: https://sqlite.org/
 [rust]: https://www.rust-lang.org/
 [swift]: https://www.swift.org/

src/chain/cbf.rs

@@ -264,6 +264,9 @@ impl CbfChainSource {
 							reorganized.len(),
 							accepted.len(),
 						);
+						*state.latest_tip.lock().unwrap() = accepted
+							.last()
+							.map(|indexed_header| indexed_header.header.block_hash());
 						let reorg_hashes = reorganized.iter().map(|h| h.block_hash()).collect();
 						state
 							.reorg_queue
@@ -398,15 +401,50 @@ impl CbfChainSource {
 				},
 			};
 
-			// Build chain checkpoint extending from the wallet's current tip.
-			let mut cp = onchain_wallet.latest_checkpoint();
-			for (height, header) in sync_update.recent_history() {
-				if *height > cp.height() {
-					let block_id = BlockId { height: *height, hash: header.block_hash() };
-					cp = cp.push(block_id).unwrap_or_else(|old| old);
-				}
-			}
 			let tip = sync_update.tip();
+			let current_cp = onchain_wallet.latest_checkpoint();
+			let reorg_detected = tip.height < current_cp.height()
+				|| sync_update.recent_history().iter().any(|(height, header)| {
+					current_cp.get(*height).is_some_and(|cp| cp.hash() != header.block_hash())
+				});
+
+			let mut cp = if reorg_detected {
+				log_debug!(
+					self.logger,
+					"Detected on-chain CBF reorg while rebuilding checkpoints at tip {}.",
+					tip.hash
+				);
+				let base_cp = sync_update
+					.recent_history()
+					.iter()
+					.find_map(|(height, header)| {
+						current_cp.get(*height).filter(|cp| cp.hash() == header.block_hash())
+					})
+					.unwrap_or_else(|| {
+						current_cp
+							.iter()
+							.last()
+							.expect("wallet checkpoint should always include genesis")
+					});
+				let mut rebuilt_cp = base_cp;
+				for (height, header) in sync_update.recent_history() {
+					if *height > rebuilt_cp.height() {
+						let block_id = BlockId { height: *height, hash: header.block_hash() };
+						rebuilt_cp = rebuilt_cp.push(block_id).unwrap_or_else(|old| old);
+					}
+				}
+				rebuilt_cp
+			} else {
+				// Build chain checkpoint extending from the wallet's current tip.
+				let mut extended_cp = current_cp;
+				for (height, header) in sync_update.recent_history() {
+					if *height > extended_cp.height() {
+						let block_id = BlockId { height: *height, hash: header.block_hash() };
+						extended_cp = extended_cp.push(block_id).unwrap_or_else(|old| old);
+					}
+				}
+				extended_cp
+			};
 			if tip.height > cp.height() {
 				let tip_block_id = BlockId { height: tip.height, hash: tip.hash };
 				cp = cp.push(tip_block_id).unwrap_or_else(|old| old);
@@ -674,28 +712,51 @@ impl CbfChainSource {
 		);
 		let fetch_start = Instant::now();
 
-		for _ in 0..FEE_RATE_LOOKBACK_BLOCKS {
+		for idx in 0..FEE_RATE_LOOKBACK_BLOCKS {
 			let remaining_timeout = timeout.saturating_sub(fetch_start.elapsed());
 			if remaining_timeout.is_zero() {
 				log_error!(self.logger, "Updating fee rate estimates timed out.");
 				return Err(Error::FeerateEstimationUpdateTimeout);
 			}
 
 			let indexed_block =
-				tokio::time::timeout(remaining_timeout, requester.get_block(current_hash))
+				match tokio::time::timeout(remaining_timeout, requester.get_block(current_hash))
 					.await
-					.map_err(|e| {
-						log_error!(self.logger, "Updating fee rate estimates timed out: {}", e);
-						Error::FeerateEstimationUpdateTimeout
-					})?
-					.map_err(|e| {
+				{
+					Ok(Ok(indexed_block)) => indexed_block,
+					Ok(Err(e)) if idx == 0 => {
+						log_debug!(
+						self.logger,
+						"Cached CBF tip {} was unavailable during fee estimation, likely due to a reorg: {:?}",
+						current_hash,
+						e
+					);
+						*self.latest_tip.lock().unwrap() = None;
+						return Ok(());
+					},
+					Ok(Err(e)) => {
 						log_error!(
 							self.logger,
 							"Failed to fetch block for fee estimation: {:?}",
 							e
 						);
-						Error::FeerateEstimationUpdateFailed
-					})?;
+						return Err(Error::FeerateEstimationUpdateFailed);
+					},
+					Err(e) if idx == 0 => {
+						log_debug!(
+						self.logger,
+						"Timed out fetching cached CBF tip {} during fee estimation, likely due to a reorg: {}",
+						current_hash,
+						e
+					);
+						*self.latest_tip.lock().unwrap() = None;
+						return Ok(());
+					},
+					Err(e) => {
+						log_error!(self.logger, "Updating fee rate estimates timed out: {}", e);
+						return Err(Error::FeerateEstimationUpdateTimeout);
+					},
+				};
 
 			let height = indexed_block.height;
 			let block = &indexed_block.block;

src/wallet/mod.rs

@@ -320,23 +320,61 @@ impl Wallet {
 
 					for mut payment in pending_payments {
 						match payment.details.kind {
-							PaymentKind::Onchain {
-								status: ConfirmationStatus::Confirmed { height, .. },
-								..
-							} => {
+							PaymentKind::Onchain { txid, .. } => {
+								let current_confirmation_status = locked_wallet
+									.tx_details(txid)
+									.map(|tx_details| match tx_details.chain_position {
+										bdk_chain::ChainPosition::Confirmed { anchor, .. } => {
+											ConfirmationStatus::Confirmed {
+												block_hash: anchor.block_id.hash,
+												height: anchor.block_id.height,
+												timestamp: anchor.confirmation_time,
+											}
+										},
+										bdk_chain::ChainPosition::Unconfirmed { .. } => {
+											ConfirmationStatus::Unconfirmed
+										},
+									});
 								let payment_id = payment.details.id;
-								if new_tip.height >= height + ANTI_REORG_DELAY - 1 {
-									payment.details.status = PaymentStatus::Succeeded;
-									self.payment_store.insert_or_update(payment.details)?;
-									self.pending_payment_store.remove(&payment_id)?;
+								match current_confirmation_status {
+									Some(ConfirmationStatus::Confirmed {
+										block_hash,
+										height,
+										timestamp,
+									}) => {
+										payment.details.kind = PaymentKind::Onchain {
+											txid,
+											status: ConfirmationStatus::Confirmed {
+												block_hash,
+												height,
+												timestamp,
+											},
+										};
+										if new_tip.height >= height + ANTI_REORG_DELAY - 1 {
+											payment.details.status = PaymentStatus::Succeeded;
+											self.payment_store.insert_or_update(payment.details)?;
+											self.pending_payment_store.remove(&payment_id)?;
+										} else {
+											self.payment_store
+												.insert_or_update(payment.details.clone())?;
+											self.pending_payment_store.insert_or_update(payment)?;
+										}
+									},
+									Some(ConfirmationStatus::Unconfirmed) | None => {
+										payment.details.kind = PaymentKind::Onchain {
+											txid,
+											status: ConfirmationStatus::Unconfirmed,
+										};
+										payment.details.status = PaymentStatus::Pending;
+										if payment.details.direction == PaymentDirection::Outbound {
+											unconfirmed_outbound_txids.push(txid);
+										}
+										self.payment_store
+											.insert_or_update(payment.details.clone())?;
+										self.pending_payment_store.insert_or_update(payment)?;
+									},
 								}
 							},
-							PaymentKind::Onchain {
-								txid,
-								status: ConfirmationStatus::Unconfirmed,
-							} if payment.details.direction == PaymentDirection::Outbound => {
-								unconfirmed_outbound_txids.push(txid);
-							},
 							_ => {},
 						}
 					}

tests/common/mod.rs

@@ -477,12 +477,9 @@ pub(crate) fn setup_node(chain_source: &TestChainSource, config: TestConfig) ->
 		TestChainSource::Cbf(bitcoind) => {
 			let p2p_socket = bitcoind.params.p2p_socket.expect("P2P must be enabled for CBF");
 			let peer_addr = format!("{}", p2p_socket);
-			let sync_config = CbfSyncConfig {
-				background_sync_config: None,
-				timeouts_config: Default::default(),
-				peers: vec![peer_addr],
-			};
-			builder.set_chain_source_cbf(Some(sync_config));
+			let sync_config =
+				CbfSyncConfig { background_sync_config: None, timeouts_config: Default::default() };
+			builder.set_chain_source_cbf(vec![peer_addr], Some(sync_config));
 		},
 	}
 

tests/integration_tests_rust.rs

@@ -21,10 +21,11 @@ use common::{
 	expect_channel_pending_event, expect_channel_ready_event, expect_channel_ready_events,
 	expect_event, expect_payment_claimable_event, expect_payment_received_event,
 	expect_payment_successful_event, expect_splice_pending_event, generate_blocks_and_wait,
-	open_channel, open_channel_push_amt, open_channel_with_all, premine_and_distribute_funds,
-	premine_blocks, prepare_rbf, random_chain_source, random_config, random_listening_addresses,
-	setup_bitcoind_and_electrsd, setup_builder, setup_node, setup_two_nodes, splice_in_with_all,
-	wait_for_tx, TestChainSource, TestStoreType, TestSyncStore,
+	invalidate_blocks, open_channel, open_channel_push_amt, open_channel_with_all,
+	premine_and_distribute_funds, premine_blocks, prepare_rbf, random_chain_source, random_config,
+	random_listening_addresses, setup_bitcoind_and_electrsd, setup_builder, setup_node,
+	setup_two_nodes, splice_in_with_all, wait_for_block, wait_for_tx, TestChainSource,
+	TestStoreType, TestSyncStore,
 };
 use ldk_node::config::{AsyncPaymentsRole, EsploraSyncConfig};
 use ldk_node::entropy::NodeEntropy;
@@ -2940,3 +2941,110 @@ async fn onchain_wallet_sync_cbf() {
 	node_a.stop().unwrap();
 	node_b.stop().unwrap();
 }
+
+#[cfg(feature = "cbf")]
+#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
+async fn onchain_wallet_sync_cbf_after_reorg() {
+	let (bitcoind, electrsd) = setup_bitcoind_and_electrsd();
+	let chain_source = TestChainSource::Cbf(&bitcoind);
+	let node = setup_node(&chain_source, random_config(true));
+
+	let first_addr = node.onchain_payment().new_address().unwrap();
+	let first_amount_sat = 100_000;
+	premine_and_distribute_funds(
+		&bitcoind.client,
+		&electrsd.client,
+		vec![first_addr],
+		Amount::from_sat(first_amount_sat),
+	)
+	.await;
+
+	wait_for_cbf_filters().await;
+	node.sync_wallets().unwrap();
+	assert_eq!(node.list_balances().spendable_onchain_balance_sats, first_amount_sat);
+
+	// Advance the tip so the reorg happens below our synced checkpoint.
+	generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 2).await;
+	wait_for_cbf_filters().await;
+	node.sync_wallets().unwrap();
+
+	// Replace the last two blocks with a different branch that has no wallet activity.
+	invalidate_blocks(&bitcoind.client, 2);
+	generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 2).await;
+	wait_for_cbf_filters().await;
+
+	let second_addr = node.onchain_payment().new_address().unwrap();
+	let second_amount_sat = 50_000;
+	distribute_funds_unconfirmed(
+		&bitcoind.client,
+		&electrsd.client,
+		vec![second_addr],
+		Amount::from_sat(second_amount_sat),
+	)
+	.await;
+	generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 1).await;
+	wait_for_cbf_filters().await;
+
+	node.sync_wallets().unwrap();
+
+	assert_eq!(
+		node.list_balances().spendable_onchain_balance_sats,
+		first_amount_sat + second_amount_sat
+	);
+	assert_eq!(
+		node.list_payments_with_filter(|p| {
+			p.direction == PaymentDirection::Inbound
+				&& matches!(p.kind, PaymentKind::Onchain { .. })
+		})
+		.len(),
+		2
+	);
+
+	node.stop().unwrap();
+}
+
+#[cfg(feature = "cbf")]
+#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
+async fn onchain_wallet_sync_cbf_reorgs_out_confirmed_receive() {
+	let (bitcoind, electrsd) = setup_bitcoind_and_electrsd();
+	let chain_source = TestChainSource::Cbf(&bitcoind);
+	let node = setup_node(&chain_source, random_config(true));
+
+	let addr = node.onchain_payment().new_address().unwrap();
+	premine_blocks(&bitcoind.client, &electrsd.client).await;
+	let cur_height = bitcoind.client.get_blockchain_info().unwrap().blocks as usize;
+	let reward_block_hash =
+		bitcoind.client.generate_to_address(1, &addr).unwrap().0.pop().unwrap().parse().unwrap();
+	wait_for_block(&electrsd.client, cur_height + 1).await;
+	let reward_block = bitcoind.client.get_block(reward_block_hash).unwrap();
+	let txid = reward_block.txdata[0].compute_txid();
+	wait_for_cbf_filters().await;
+
+	node.sync_wallets().unwrap();
+	assert_eq!(node.list_balances().total_onchain_balance_sats, 5_000_000_000);
+	assert_eq!(node.list_balances().spendable_onchain_balance_sats, 0);
+
+	let payment_id = PaymentId(txid.to_byte_array());
+	let payment = node.payment(&payment_id).unwrap();
+	assert_eq!(payment.status, PaymentStatus::Pending);
+	match payment.kind {
+		PaymentKind::Onchain { status: ConfirmationStatus::Confirmed { .. }, .. } => {},
+		other => panic!("Unexpected payment state before reorg: {:?}", other),
+	}
+
+	invalidate_blocks(&bitcoind.client, 1);
+	generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 2).await;
+	wait_for_cbf_filters().await;
+	node.sync_wallets().unwrap();
+
+	assert_eq!(node.list_balances().total_onchain_balance_sats, 0);
+	assert_eq!(node.list_balances().spendable_onchain_balance_sats, 0);
+	let payment = node.payment(&payment_id).unwrap();
+	assert_eq!(payment.status, PaymentStatus::Pending);
+	match payment.kind {
+		PaymentKind::Onchain { status: ConfirmationStatus::Unconfirmed, .. } => {},
+		other => panic!("Unexpected payment state after reorg: {:?}", other),
+	}
+
+	node.stop().unwrap();
+}