Pass a BestBlock to init::synchronize_listeners

On restart, LDK expects the chain to be replayed starting from
where it was when objects were last serialized. This is fine in the
normal case, but if there was a reorg and the node which we were
syncing from either resynced or was changed, the last block that we
were synced as of might no longer be available. As a result, it
becomes impossible to figure out where the fork point is, and thus
to replay the chain.

Luckily, changing the block source during a reorg isn't exactly
common, but we shouldn't end up with a bricked node.

To address this, `lightning-block-sync` allows the user to pass in
`Cache` which can be used to cache recent blocks and thus allow for
reorg handling in this case. However, serialization for, and a
reasonable default implementation of a `Cache` was never built.

Instead, here, we start taking a different approach. To avoid
developers having to persist yet another object, we move
`BestBlock` to storing some number of recent block hashes. This
allows us to find the fork point with just the serialized state.

In a previous commit, we moved deserialization of various structs
to return the `BestBlock` rather than a `BlockHash`. Here we move
to actually using it, taking a `BestBlock` in place of `BlockHash`
to `init::synchronize_listeners` and walking the `previous_blocks`
list to find the fork point rather than relying on the `Cache`.

Author: Matt Corallo

lightning-block-sync/src/init.rs

@@ -117,8 +117,8 @@ where
 /// 	let mut cache = UnboundedCache::new();
 /// 	let mut monitor_listener = (monitor, &*tx_broadcaster, &*fee_estimator, &*logger);
 /// 	let listeners = vec![
-/// 		(monitor_best_block.block_hash, &monitor_listener as &dyn chain::Listen),
-/// 		(manager_best_block.block_hash, &manager as &dyn chain::Listen),
+/// 		(monitor_best_block, &monitor_listener as &dyn chain::Listen),
+/// 		(manager_best_block, &manager as &dyn chain::Listen),
 /// 	];
 /// 	let chain_tip = init::synchronize_listeners(
 /// 		block_source, Network::Bitcoin, &mut cache, listeners).await.unwrap();
@@ -143,39 +143,28 @@ pub async fn synchronize_listeners<
 	L: chain::Listen + ?Sized,
 >(
 	block_source: B, network: Network, header_cache: &mut C,
-	mut chain_listeners: Vec<(BlockHash, &L)>,
+	mut chain_listeners: Vec<(BestBlock, &L)>,
 ) -> BlockSourceResult<ValidatedBlockHeader>
 where
 	B::Target: BlockSource,
 {
 	let best_header = validate_best_block_header(&*block_source).await?;
 
-	// Fetch the header for the block hash paired with each listener.
-	let mut chain_listeners_with_old_headers = Vec::new();
-	for (old_block_hash, chain_listener) in chain_listeners.drain(..) {
-		let old_header = match header_cache.look_up(&old_block_hash) {
-			Some(header) => *header,
-			None => {
-				block_source.get_header(&old_block_hash, None).await?.validate(old_block_hash)?
-			},
-		};
-		chain_listeners_with_old_headers.push((old_header, chain_listener))
-	}
-
 	// Find differences and disconnect blocks for each listener individually.
 	let mut chain_poller = ChainPoller::new(block_source, network);
 	let mut chain_listeners_at_height = Vec::new();
 	let mut most_common_ancestor = None;
 	let mut most_connected_blocks = Vec::new();
-	for (old_header, chain_listener) in chain_listeners_with_old_headers.drain(..) {
+	for (old_best_block, chain_listener) in chain_listeners.drain(..) {
 		// Disconnect any stale blocks, but keep them in the cache for the next iteration.
 		let header_cache = &mut ReadOnlyCache(header_cache);
 		let (common_ancestor, connected_blocks) = {
 			let chain_listener = &DynamicChainListener(chain_listener);
 			let mut chain_notifier = ChainNotifier { header_cache, chain_listener };
-			let difference =
-				chain_notifier.find_difference(best_header, &old_header, &mut chain_poller).await?;
-			if difference.common_ancestor != old_header {
+			let difference = chain_notifier
+				.find_difference_from_best_block(best_header, old_best_block, &mut chain_poller)
+				.await?;
+			if difference.common_ancestor.block_hash != old_best_block.block_hash {
 				chain_notifier.disconnect_blocks(difference.common_ancestor);
 			}
 			(difference.common_ancestor, difference.connected_blocks)
@@ -281,9 +270,9 @@ mod tests {
 		let listener_3 = MockChainListener::new().expect_block_connected(*chain.at_height(4));
 
 		let listeners = vec![
-			(chain.at_height(1).block_hash, &listener_1 as &dyn chain::Listen),
-			(chain.at_height(2).block_hash, &listener_2 as &dyn chain::Listen),
-			(chain.at_height(3).block_hash, &listener_3 as &dyn chain::Listen),
+			(chain.best_block_at_height(1), &listener_1 as &dyn chain::Listen),
+			(chain.best_block_at_height(2), &listener_2 as &dyn chain::Listen),
+			(chain.best_block_at_height(3), &listener_3 as &dyn chain::Listen),
 		];
 		let mut cache = chain.header_cache(0..=4);
 		match synchronize_listeners(&chain, Network::Bitcoin, &mut cache, listeners).await {
@@ -313,9 +302,9 @@ mod tests {
 			.expect_block_connected(*main_chain.at_height(4));
 
 		let listeners = vec![
-			(fork_chain_1.tip().block_hash, &listener_1 as &dyn chain::Listen),
-			(fork_chain_2.tip().block_hash, &listener_2 as &dyn chain::Listen),
-			(fork_chain_3.tip().block_hash, &listener_3 as &dyn chain::Listen),
+			(fork_chain_1.best_block(), &listener_1 as &dyn chain::Listen),
+			(fork_chain_2.best_block(), &listener_2 as &dyn chain::Listen),
+			(fork_chain_3.best_block(), &listener_3 as &dyn chain::Listen),
 		];
 		let mut cache = fork_chain_1.header_cache(2..=4);
 		cache.extend(fork_chain_2.header_cache(3..=4));
@@ -350,9 +339,9 @@ mod tests {
 			.expect_block_connected(*main_chain.at_height(4));
 
 		let listeners = vec![
-			(fork_chain_1.tip().block_hash, &listener_1 as &dyn chain::Listen),
-			(fork_chain_2.tip().block_hash, &listener_2 as &dyn chain::Listen),
-			(fork_chain_3.tip().block_hash, &listener_3 as &dyn chain::Listen),
+			(fork_chain_1.best_block(), &listener_1 as &dyn chain::Listen),
+			(fork_chain_2.best_block(), &listener_2 as &dyn chain::Listen),
+			(fork_chain_3.best_block(), &listener_3 as &dyn chain::Listen),
 		];
 		let mut cache = fork_chain_1.header_cache(2..=4);
 		cache.extend(fork_chain_2.header_cache(3..=4));
@@ -368,18 +357,18 @@ mod tests {
 		let main_chain = Blockchain::default().with_height(2);
 		let fork_chain = main_chain.fork_at_height(1);
 		let new_tip = main_chain.tip();
-		let old_tip = fork_chain.tip();
+		let old_best_block = fork_chain.best_block();
 
 		let listener = MockChainListener::new()
 			.expect_blocks_disconnected(*fork_chain.at_height(1))
 			.expect_block_connected(*new_tip);
 
-		let listeners = vec![(old_tip.block_hash, &listener as &dyn chain::Listen)];
+		let listeners = vec![(old_best_block, &listener as &dyn chain::Listen)];
 		let mut cache = fork_chain.header_cache(2..=2);
 		match synchronize_listeners(&main_chain, Network::Bitcoin, &mut cache, listeners).await {
 			Ok(_) => {
 				assert!(cache.contains_key(&new_tip.block_hash));
-				assert!(cache.contains_key(&old_tip.block_hash));
+				assert!(cache.contains_key(&old_best_block.block_hash));
 			},
 			Err(e) => panic!("Unexpected error: {:?}", e),
 		}

lightning-block-sync/src/lib.rs

@@ -337,7 +337,7 @@ where
 		chain_poller: &mut P,
 	) -> Result<(), (BlockSourceError, Option<ValidatedBlockHeader>)> {
 		let difference = self
-			.find_difference(new_header, old_header, chain_poller)
+			.find_difference_from_header(new_header, old_header, chain_poller)
 			.await
 			.map_err(|e| (e, None))?;
 		if difference.common_ancestor != *old_header {
@@ -347,11 +347,52 @@ where
 			.await
 	}
 
+	/// Returns the changes needed to produce the chain with `current_header` as its tip from the
+	/// chain with `prev_best_block` as its tip.
+	///
+	/// First resolves `prev_best_block` to a `ValidatedBlockHeader` using the `previous_blocks`
+	/// field as fallback if needed, then finds the common ancestor.
+	async fn find_difference_from_best_block<P: Poll>(
+		&self, current_header: ValidatedBlockHeader, prev_best_block: BestBlock,
+		chain_poller: &mut P,
+	) -> BlockSourceResult<ChainDifference> {
+		// Try to resolve the header for the previous best block. First try the block_hash,
+		// then fall back to previous_blocks if that fails.
+		let cur_tip = core::iter::once((0, &prev_best_block.block_hash));
+		let prev_tips =
+			prev_best_block.previous_blocks.iter().enumerate().filter_map(|(idx, hash_opt)| {
+				if let Some(block_hash) = hash_opt {
+					Some((idx as u32 + 1, block_hash))
+				} else {
+					None
+				}
+			});
+		let mut found_header = None;
+		for (height_diff, block_hash) in cur_tip.chain(prev_tips) {
+			if let Some(header) = self.header_cache.look_up(block_hash) {
+				found_header = Some(*header);
+				break;
+			}
+			let height = prev_best_block.height.checked_sub(height_diff).ok_or(
+				BlockSourceError::persistent("BestBlock had more previous_blocks than its height"),
+			)?;
+			if let Ok(header) = chain_poller.get_header(block_hash, Some(height)).await {
+				found_header = Some(header);
+				break;
+			}
+		}
+		let found_header = found_header.ok_or_else(|| {
+			BlockSourceError::persistent("could not resolve any block from BestBlock")
+		})?;
+
+		self.find_difference_from_header(current_header, &found_header, chain_poller).await
+	}
+
 	/// Returns the changes needed to produce the chain with `current_header` as its tip from the
 	/// chain with `prev_header` as its tip.
 	///
 	/// Walks backwards from `current_header` and `prev_header`, finding the common ancestor.
-	async fn find_difference<P: Poll>(
+	async fn find_difference_from_header<P: Poll>(
 		&self, current_header: ValidatedBlockHeader, prev_header: &ValidatedBlockHeader,
 		chain_poller: &mut P,
 	) -> BlockSourceResult<ChainDifference> {

lightning-block-sync/src/poll.rs

@@ -31,6 +31,11 @@ pub trait Poll {
 	fn fetch_block<'a>(
 		&'a self, header: &'a ValidatedBlockHeader,
 	) -> impl Future<Output = BlockSourceResult<ValidatedBlock>> + Send + 'a;
+
+	/// Returns the header for a given hash and optional height hint.
+	fn get_header<'a>(
+		&'a self, block_hash: &'a BlockHash, height_hint: Option<u32>,
+	) -> impl Future<Output = BlockSourceResult<ValidatedBlockHeader>> + Send + 'a;
 }
 
 /// A chain tip relative to another chain tip in terms of block hash and chainwork.
@@ -258,6 +263,14 @@ impl<B: Deref<Target = T> + Sized + Send + Sync, T: BlockSource + ?Sized> Poll
 	) -> impl Future<Output = BlockSourceResult<ValidatedBlock>> + Send + 'a {
 		async move { self.block_source.get_block(&header.block_hash).await?.validate(header.block_hash) }
 	}
+
+	fn get_header<'a>(
+		&'a self, block_hash: &'a BlockHash, height_hint: Option<u32>,
+	) -> impl Future<Output = BlockSourceResult<ValidatedBlockHeader>> + Send + 'a {
+		Box::pin(async move {
+			self.block_source.get_header(block_hash, height_hint).await?.validate(*block_hash)
+		})
+	}
 }
 
 #[cfg(test)]

lightning-block-sync/src/test_utils.rs

@@ -104,6 +104,18 @@ impl Blockchain {
 		block_header.validate(block_hash).unwrap()
 	}
 
+	pub fn best_block_at_height(&self, height: usize) -> BestBlock {
+		let mut previous_blocks = [None; 12];
+		for (i, height) in (0..height).rev().take(12).enumerate() {
+			previous_blocks[i] = Some(self.blocks[height].block_hash());
+		}
+		BestBlock {
+			height: height as u32,
+			block_hash: self.blocks[height].block_hash(),
+			previous_blocks,
+		}
+	}
+
 	fn at_height_unvalidated(&self, height: usize) -> BlockHeaderData {
 		assert!(!self.blocks.is_empty());
 		assert!(height < self.blocks.len());
@@ -123,6 +135,11 @@ impl Blockchain {
 		self.at_height(self.blocks.len() - 1)
 	}
 
+	pub fn best_block(&self) -> BestBlock {
+		assert!(!self.blocks.is_empty());
+		self.best_block_at_height(self.blocks.len() - 1)
+	}
+
 	pub fn disconnect_tip(&mut self) -> Option<Block> {
 		self.blocks.pop()
 	}