Include recent blocks in the synchronize_listeners-returned cache

When `synchronize_listeners` runs, it returns a cache of the
headers it needed when doing chain difference-finding. This allows
us to ensure that when we start running normally we have all the
recent headers in case we need them to reorg.

Sadly, in some cases it was returning a mostly-empty cache.
Because it was only being filled during block difference
reconciliation it would only get a block around each listener's
fork point. Worse, because we were calling `disconnect_blocks` with
the cache the cache would assume we were reorging against the main
chain and drop blocks we actually want.

Instead, we avoid dropping blocks on `disconnect_blocks` calls and
ensure we always add connected blocks to the cache.

Author: Matt Corallo

lightning-block-sync/src/init.rs

@@ -152,6 +152,7 @@ where
 	let mut chain_listeners_at_height = Vec::new();
 	let mut most_connected_blocks = Vec::new();
 	let mut header_cache = HeaderCache::new();
+	header_cache.retain_on_disconnect = true;
 	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 (common_ancestor, connected_blocks) = {
@@ -192,7 +193,9 @@ where
 		const NO_BLOCK: Option<(u32, crate::poll::ValidatedBlock)> = None;
 		let mut fetched_blocks = [NO_BLOCK; MAX_BLOCKS_AT_ONCE];
 		for ((header, block_res), result) in results.into_iter().zip(fetched_blocks.iter_mut()) {
-			*result = Some((header.height, block_res?));
+			let block = block_res?;
+			header_cache.block_connected(header.block_hash, *header);
+			*result = Some((header.height, block));
 		}
 		debug_assert!(fetched_blocks.iter().take(most_connected_blocks.len()).all(|r| r.is_some()));
 		// TODO: When our MSRV is 1.82, use is_sorted_by_key
@@ -225,6 +228,7 @@ where
 			.truncate(most_connected_blocks.len().saturating_sub(MAX_BLOCKS_AT_ONCE));
 	}
 
+	header_cache.retain_on_disconnect = false;
 	Ok((header_cache, best_header))
 }
 
@@ -267,7 +271,13 @@ mod tests {
 			(chain.best_block_at_height(3), &listener_3 as &dyn chain::Listen),
 		];
 		match synchronize_listeners(&chain, Network::Bitcoin, listeners).await {
-			Ok((_, header)) => assert_eq!(header, chain.tip()),
+			Ok((cache, header)) => {
+				assert_eq!(header, chain.tip());
+				assert!(cache.look_up(&chain.at_height(1).block_hash).is_some());
+				assert!(cache.look_up(&chain.at_height(2).block_hash).is_some());
+				assert!(cache.look_up(&chain.at_height(3).block_hash).is_some());
+				assert!(cache.look_up(&chain.at_height(4).block_hash).is_some());
+			},
 			Err(e) => panic!("Unexpected error: {:?}", e),
 		}
 	}
@@ -298,7 +308,15 @@ mod tests {
 			(fork_chain_3.best_block(), &listener_3 as &dyn chain::Listen),
 		];
 		match synchronize_listeners(&main_chain, Network::Bitcoin, listeners).await {
-			Ok((_, header)) => assert_eq!(header, main_chain.tip()),
+			Ok((cache, header)) => {
+				assert_eq!(header, main_chain.tip());
+				assert!(cache.look_up(&main_chain.at_height(1).block_hash).is_some());
+				assert!(cache.look_up(&main_chain.at_height(2).block_hash).is_some());
+				assert!(cache.look_up(&main_chain.at_height(3).block_hash).is_some());
+				assert!(cache.look_up(&fork_chain_1.at_height(2).block_hash).is_none());
+				assert!(cache.look_up(&fork_chain_2.at_height(3).block_hash).is_none());
+				assert!(cache.look_up(&fork_chain_3.at_height(4).block_hash).is_none());
+			},
 			Err(e) => panic!("Unexpected error: {:?}", e),
 		}
 	}
@@ -332,7 +350,16 @@ mod tests {
 			(fork_chain_3.best_block(), &listener_3 as &dyn chain::Listen),
 		];
 		match synchronize_listeners(&main_chain, Network::Bitcoin, listeners).await {
-			Ok((_, header)) => assert_eq!(header, main_chain.tip()),
+			Ok((cache, header)) => {
+				assert_eq!(header, main_chain.tip());
+				assert!(cache.look_up(&main_chain.at_height(1).block_hash).is_some());
+				assert!(cache.look_up(&main_chain.at_height(2).block_hash).is_some());
+				assert!(cache.look_up(&main_chain.at_height(3).block_hash).is_some());
+				assert!(cache.look_up(&main_chain.at_height(4).block_hash).is_some());
+				assert!(cache.look_up(&fork_chain_1.at_height(2).block_hash).is_none());
+				assert!(cache.look_up(&fork_chain_1.at_height(3).block_hash).is_none());
+				assert!(cache.look_up(&fork_chain_1.at_height(4).block_hash).is_none());
+			},
 			Err(e) => panic!("Unexpected error: {:?}", e),
 		}
 	}

lightning-block-sync/src/lib.rs

@@ -193,12 +193,15 @@ pub const HEADER_CACHE_LIMIT: u32 = 6 * 24 * 7;
 /// Retains only the latest [`HEADER_CACHE_LIMIT`] block headers based on height.
 pub struct HeaderCache {
 	headers: std::collections::HashMap<BlockHash, ValidatedBlockHeader>,
+	/// When set, [`Self::blocks_disconnected`] will not evict headers above the fork point.
+	/// This is used during initial sync to retain headers across multiple listeners.
+	retain_on_disconnect: bool,
 }
 
 impl HeaderCache {
 	/// Creates a new empty header cache.
 	pub fn new() -> Self {
-		Self { headers: std::collections::HashMap::new() }
+		Self { headers: std::collections::HashMap::new(), retain_on_disconnect: false }
 	}
 
 	/// Retrieves the block header keyed by the given block hash.
@@ -234,9 +237,12 @@ impl HeaderCache {
 	/// Called when blocks have been disconnected from the best chain. Only the fork point
 	/// (best common ancestor) is provided.
 	///
-	/// Once disconnected, a block's header is no longer needed and thus can be removed.
+	/// Once disconnected, unless [`Self::retain_on_disconnect`] is set, a block's header is no
+	/// longer needed and thus can be removed.
 	pub(crate) fn blocks_disconnected(&mut self, fork_point: &ValidatedBlockHeader) {
-		self.headers.retain(|_, block_info| block_info.height <= fork_point.height);
+		if !self.retain_on_disconnect {
+			self.headers.retain(|_, block_info| block_info.height <= fork_point.height);
+		}
 	}
 }