cbf: add preprocessing step before kyoto runs

A node might go offline on a reorganized block, which would make
syncing stall after a restart.

This commit adds a preprocessing step which runs once before the events
from kyoto are processed.

During the preprocessing step we check whether we have forked and
disconnect blocks accordingly. Add a test for a deep reorg that happens
while the node is offline.

Still there might be a problem if several of chain monitors diverged and
live on different forks, but other chain sources (bitcoind) ignore that
and that perhaps requires a fix upstream.

Author: randomlogin

src/chain/cbf.rs

@@ -6,8 +6,8 @@ use std::time::{Duration, SystemTime, UNIX_EPOCH};
 use bip157::chain::ChainState;
 use bip157::{
 	chain::BlockHeaderChanges, Builder as KyotoBuilder, Client, Event as KyotoEvent,
-	HashCheckpoint, Header, IndexedBlock, Info, Node as KyotoNode, Package, Requester, TrustedPeer,
-	Warning,
+	HashCheckpoint, Header, IndexedBlock, Info, Node as KyotoNode, Package,
+	Requester as KyotoRequester, TrustedPeer, Warning,
 };
 use bitcoin::{BlockHash, FeeRate, Network, Script, ScriptBuf, Transaction, Txid};
 use electrum_client::{Client as ElectrumClient, ConfigBuilder as ElectrumConfigBuilder};
@@ -60,7 +60,7 @@ const ELECTRUM_FEE_TIMEOUT_SECS: u64 = 10;
 
 /// Runtime status of the underlying kyoto node.
 enum CbfRuntimeStatus {
-	Started { requester: Requester },
+	Started { requester: KyotoRequester },
 	Stopped,
 }
 
@@ -182,9 +182,10 @@ impl BlockApplicator {
 				},
 				ChainOp::Disconnect { fork_point } => {
 					self.chain_listener.blocks_disconnected(fork_point);
-					self.next_height = fork_point.height + 1;
+					let best_block_height = self.chain_listener.get_best_block().height;
+					self.next_height = best_block_height + 1;
 					self.sync_state_tx.send_replace(CbfSyncState::Active {
-						applied_tip: Some(fork_point.height),
+						applied_tip: Some(best_block_height),
 						synced_to_tip: false,
 					});
 				},
@@ -430,9 +431,19 @@ impl CbfChainSource {
 					Arc::clone(&restart_logger),
 				));
 
+				let (preprocessed_event_tx, preprocessed_event_rx) =
+					mpsc::unbounded_channel::<KyotoEvent>();
+				let _ = tokio::spawn(Self::preprocess_kyoto_events(
+					current_event_rx,
+					preprocessed_event_tx,
+					restart_listener.clone(),
+					Arc::clone(&restart_cbf_runtime_status),
+					ops_tx.clone(),
+				));
+
 				let event_handle = tokio::spawn(Self::process_kyoto_events(
 					Arc::clone(&restart_logger),
-					current_event_rx,
+					preprocessed_event_rx,
 					Arc::clone(&restart_registered_scripts),
 					Arc::clone(&restart_cbf_runtime_status),
 					ops_tx.clone(),
@@ -583,6 +594,39 @@ impl CbfChainSource {
 		}
 	}
 
+	//this function is called once per start before we really process the events
+	//we need it to decide if we have been reorganized during the time we were offline
+	async fn preprocess_kyoto_events(
+		mut kyoto_events_rx: mpsc::UnboundedReceiver<KyotoEvent>,
+		processed_events_tx: mpsc::UnboundedSender<KyotoEvent>, chain_listener: ChainListener,
+		cbf_runtime_status: Arc<Mutex<CbfRuntimeStatus>>, ops_tx: mpsc::UnboundedSender<ChainOp>,
+	) {
+		let mut checked_for_offline_reorg = false;
+		while let Some(event) = kyoto_events_rx.recv().await {
+			if !checked_for_offline_reorg {
+				match &event {
+					// if we received filter, we're in canonical chain and can check for forks occured
+					KyotoEvent::IndexedFilter(_) => {
+						checked_for_offline_reorg = true;
+						let requester = match &*cbf_runtime_status.lock().expect("lock") {
+							CbfRuntimeStatus::Started { requester } => requester.clone(),
+							CbfRuntimeStatus::Stopped => return,
+						};
+						if let Some(fork_point) =
+							startup_fork_point(&requester, &chain_listener).await
+						{
+							let _ = ops_tx.send(ChainOp::Disconnect { fork_point });
+						}
+					},
+					_ => {},
+				}
+			}
+			if processed_events_tx.send(event).is_err() {
+				return;
+			}
+		}
+	}
+
 	async fn process_kyoto_events(
 		logger: Arc<Logger>, mut event_rx: mpsc::UnboundedReceiver<KyotoEvent>,
 		registered_scripts: Arc<Mutex<HashSet<ScriptBuf>>>,
@@ -930,7 +974,7 @@ impl CbfChainSource {
 	/// means we have no recent data yet. The window therefore fills incrementally over successive
 	/// updates rather than requiring all [`FEE_WINDOW_BLOCKS`] downloads to succeed at once.
 	async fn refresh_block_fee_window(
-		&self, requester: &Requester, cache: &Mutex<BTreeMap<u32, (BlockHash, FeeRate)>>,
+		&self, requester: &KyotoRequester, cache: &Mutex<BTreeMap<u32, (BlockHash, FeeRate)>>,
 	) -> Vec<FeeRate> {
 		let tip_height = match requester.chain_tip().await {
 			Ok(tip) => tip.height,
@@ -1003,6 +1047,48 @@ impl CbfChainSource {
 	}
 }
 
+///Fork point which might have occured when we were offline
+async fn startup_fork_point(
+	requester: &KyotoRequester, chain_listener: &ChainListener,
+) -> Option<BlockLocator> {
+	let canonical_tip_height = requester.chain_tip().await.ok()?.height;
+
+	let mut fork_point: Option<BlockLocator> = None;
+	for best_block in chain_listener.best_blocks() {
+		// The consumer's tip followed by the hashes of the blocks right below it.
+		let mut known_blocks = vec![(best_block.height, best_block.block_hash)];
+		for (i, previous_hash) in best_block.previous_blocks.iter().enumerate() {
+			match (best_block.height.checked_sub(i as u32 + 1), previous_hash) {
+				(Some(height), Some(hash)) => known_blocks.push((height, *hash)),
+				_ => break,
+			}
+		}
+
+		for (height, hash) in known_blocks {
+			if height > canonical_tip_height {
+				continue;
+			}
+			let canonical_header = match requester.get_header(height).await.ok().flatten() {
+				Some(header) => header,
+				// Kyoto has no headers below its checkpoint.
+				None => break,
+			};
+			if canonical_header.block_hash() != hash {
+				continue;
+			}
+			// The consumer's last canonical block. If it is its tip, nothing needs disconnecting.
+			if height < best_block.height {
+				match fork_point {
+					Some(lowest) if lowest.height <= height => {},
+					_ => fork_point = Some(BlockLocator::new(hash, height)),
+				}
+			}
+			break;
+		}
+	}
+	fork_point
+}
+
 fn resume_checkpoint(logger: &Logger, chain_listener: &ChainListener) -> Option<HashCheckpoint> {
 	let min_best_block = chain_listener.get_best_block();
 	let bdk_cp = chain_listener.onchain_wallet.upgrade()?.latest_checkpoint();

src/chain/mod.rs

@@ -21,9 +21,9 @@ use std::sync::{Arc, Mutex};
 use std::time::Duration;
 
 use bitcoin::{Script, Txid};
-use lightning::chain::{BlockLocator, Filter};
 #[cfg(any(feature = "chain-bitcoind", feature = "chain-cbf"))]
 use lightning::chain::Listen;
+use lightning::chain::{BlockLocator, Filter};
 
 #[cfg(feature = "chain-bitcoind")]
 use crate::chain::bitcoind::{BitcoindChainSource, UtxoSourceClient};
@@ -207,27 +207,26 @@ impl ChainListener {
 
 #[cfg(any(feature = "chain-bitcoind", feature = "chain-cbf"))]
 impl ChainListener {
-	pub(crate) fn get_best_block(&self) -> BlockLocator {
+	pub(crate) fn best_blocks(&self) -> Vec<BlockLocator> {
 		let (onchain_wallet, channel_manager, chain_monitor, output_sweeper) =
 			self.upgrade().expect("chain listener components dropped");
-		let candidates = [
+		let mut best_blocks = vec![
 			onchain_wallet.current_best_block(),
 			channel_manager.current_best_block(),
 			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) = chain_monitor
-			.list_monitors()
-			.iter()
-			.flat_map(|id| 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
+		best_blocks.extend(
+			chain_monitor
+				.list_monitors()
+				.iter()
+				.flat_map(|id| chain_monitor.get_monitor(*id))
+				.map(|m| m.current_best_block()),
+		);
+		best_blocks
+	}
+
+	pub(crate) fn get_best_block(&self) -> BlockLocator {
+		self.best_blocks().into_iter().min_by_key(|b| b.height).expect("non-empty")
 	}
 }
 
@@ -283,7 +282,20 @@ impl Listen for ChainListener {
 			if channel_manager.current_best_block().height > fork_point_block.height {
 				channel_manager.blocks_disconnected(fork_point_block);
 			}
-			chain_monitor.blocks_disconnected(fork_point_block);
+			// Monitors are persisted at different heights, and `ChainMonitor::blocks_disconnected`
+			// asserts every one of them is above the fork point, so disconnect them one by one.
+			for monitor in
+				chain_monitor.list_monitors().iter().flat_map(|id| chain_monitor.get_monitor(*id))
+			{
+				if monitor.current_best_block().height > fork_point_block.height {
+					monitor.blocks_disconnected(
+						fork_point_block,
+						onchain_wallet.broadcaster(),
+						onchain_wallet.fee_estimator(),
+						onchain_wallet.logger(),
+					);
+				}
+			}
 			if output_sweeper.current_best_block().height > fork_point_block.height {
 				output_sweeper.blocks_disconnected(fork_point_block);
 			}
@@ -730,9 +742,7 @@ impl ChainSource {
 				bitcoind_chain_source.update_fee_rate_estimates().await
 			},
 			#[cfg(feature = "chain-cbf")]
-			ChainSourceKind::Cbf(cbf_chain_source) => {
-				cbf_chain_source.update_fee_rate_estimates().await
-			},
+			ChainSourceKind::Cbf(cbf_chain_source) => cbf_chain_source.update_fee_rate_estimates().await,
 		}
 	}
 

src/wallet/mod.rs

@@ -235,6 +235,21 @@ impl Wallet {
 		self.inner.lock().expect("lock").latest_checkpoint()
 	}
 
+	#[cfg(any(feature = "chain-bitcoind", feature = "chain-cbf"))]
+	pub(crate) fn broadcaster(&self) -> &Arc<Broadcaster> {
+		&self.broadcaster
+	}
+
+	#[cfg(any(feature = "chain-bitcoind", feature = "chain-cbf"))]
+	pub(crate) fn fee_estimator(&self) -> &Arc<OnchainFeeEstimator> {
+		&self.fee_estimator
+	}
+
+	#[cfg(any(feature = "chain-bitcoind", feature = "chain-cbf"))]
+	pub(crate) fn logger(&self) -> &Arc<Logger> {
+		&self.logger
+	}
+
 	#[cfg(any(feature = "chain-bitcoind", feature = "chain-cbf"))]
 	pub(crate) fn current_best_block(&self) -> BlockLocator {
 		let checkpoint = self.inner.lock().expect("lock").latest_checkpoint();
@@ -469,8 +484,12 @@ impl Wallet {
 											let current = existing?;
 											match current.kind {
 												PaymentKind::Onchain {
-													status: ConfirmationStatus::Confirmed { height, .. }, ..
-												} if new_tip.height >= height + ANTI_REORG_DELAY - 1 => {
+													status:
+														ConfirmationStatus::Confirmed { height, .. },
+													..
+												} if new_tip.height
+													>= height + ANTI_REORG_DELAY - 1 =>
+												{
 													graduated = true;
 													let mut update =
 														PaymentDetailsUpdate::new(payment_id);
@@ -527,7 +546,8 @@ impl Wallet {
 							let mut locked_wallet = self.inner.lock().expect("lock");
 							locked_wallet.apply_evicted_txs_events(evicted_txids)
 						};
-						Box::pin(self.update_payment_store(eviction_events, reference_time)).await?;
+						Box::pin(self.update_payment_store(eviction_events, reference_time))
+							.await?;
 					}
 
 					if !unconfirmed_outbound_txids.is_empty() {
@@ -4576,9 +4596,9 @@ mod tests {
 			wallet.update_payment_store(vec![event], reference_time),
 		)
 		.await;
-		assert!(
-			result.expect("eviction must not wait on the persister lock held by its caller").is_ok()
-		);
+		assert!(result
+			.expect("eviction must not wait on the persister lock held by its caller")
+			.is_ok());
 
 		let payment = wallet.payment_store.get(&payment_id).await.unwrap().unwrap();
 		assert_eq!(payment.status, PaymentStatus::Failed);

tests/reorg_test.rs

@@ -12,8 +12,8 @@ use serde_json::json;
 use crate::common::{
 	expect_event, exponential_backoff_poll, generate_blocks_and_wait, invalidate_blocks,
 	open_channel, premine_and_distribute_funds, random_chain_source, random_config,
-	setup_bitcoind_and_electrsd, setup_node, wait_for_outpoint_spend, wait_for_tx, NodePaymentExt,
-	TestChainSource,
+	setup_bitcoind_and_electrsd, setup_node, wait_for_node_tip, wait_for_outpoint_spend,
+	wait_for_tx, NodePaymentExt, TestChainSource, TestStoreType,
 };
 
 #[cfg(feature = "chain-bitcoind")]
@@ -72,6 +72,41 @@ fn bitcoind_rest_follows_valid_reorg() {
 	})
 }
 
+#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
+async fn restart_follows_deep_reorg_that_happened_while_offline() {
+	let (bitcoind, electrsd) = setup_bitcoind_and_electrsd();
+	let chain_source = random_chain_source(&bitcoind, &electrsd);
+	let mut config = random_config();
+	config.store_type = TestStoreType::Sqlite;
+	let node = setup_node(&chain_source, config.clone());
+
+	generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 50).await;
+	node.sync_wallets().unwrap();
+	let old_tip = node.status().current_best_block;
+
+	node.stop().unwrap();
+	drop(node);
+
+	let reorg_depth = 5;
+	invalidate_blocks(&bitcoind.client, reorg_depth);
+	let new_tip_height =
+		generate_blocks_and_wait(&bitcoind.client, &electrsd.client, reorg_depth + 20).await;
+	let new_tip_hash = bitcoind.client.best_block_hash().expect("failed to get new tip hash");
+	assert_ne!(old_tip.block_hash, new_tip_hash);
+
+	let node = setup_node(&chain_source, config);
+	wait_for_node_tip(&node, new_tip_height)
+		.await
+		.expect("restarted node did not reach the new tip after an offline reorg");
+	node.sync_wallets().unwrap();
+
+	let synced_tip = node.status().current_best_block;
+	assert_eq!(synced_tip.height as usize, new_tip_height);
+	assert_eq!(synced_tip.block_hash, new_tip_hash);
+
+	node.stop().unwrap();
+}
+
 async fn wait_for_pending_sweep_balance<F>(
 	node: &ldk_node::Node, mut matches_balance: F,
 ) -> PendingSweepBalance