Make the Cache trait priv, just use UnboundedCache publicly

In the previous commit, we moved to relying on
`BestBlock::previous_blocks` to find the fork point in
`lightning-block-sync`'s `init::synchronize_listeners`. Here we now
drop the `Cache` parameter as we no longer rely on it.

Because we now have no reason to want a persistent `Cache`, we
remove the trait from the public interface. However, to keep
disconnections reliable we return the `UnboundedCache` we built up
during initial sync from `init::synchronize_listeners` which we
expect developers to pass to `SpvClient::new`.

Author: Matt Corallo

lightning-block-sync/src/init.rs

@@ -2,7 +2,7 @@
 //! from disk.
 
 use crate::poll::{ChainPoller, Validate, ValidatedBlockHeader};
-use crate::{BlockSource, BlockSourceResult, Cache, ChainNotifier};
+use crate::{BlockSource, BlockSourceResult, Cache, ChainNotifier, UnboundedCache};
 
 use bitcoin::block::Header;
 use bitcoin::hash_types::BlockHash;
@@ -32,9 +32,9 @@ where
 /// Performs a one-time sync of chain listeners using a single *trusted* block source, bringing each
 /// listener's view of the chain from its paired block hash to `block_source`'s best chain tip.
 ///
-/// Upon success, the returned header can be used to initialize [`SpvClient`]. In the case of
-/// failure, each listener may be left at a different block hash than the one it was originally
-/// paired with.
+/// Upon success, the returned header and header cache can be used to initialize [`SpvClient`]. In
+/// the case of failure, each listener may be left at a different block hash than the one it was
+/// originally paired with.
 ///
 /// Useful during startup to bring the [`ChannelManager`] and each [`ChannelMonitor`] in sync before
 /// switching to [`SpvClient`]. For example:
@@ -114,14 +114,13 @@ where
 /// 	};
 ///
 /// 	// Synchronize any channel monitors and the channel manager to be on the best block.
-/// 	let mut cache = UnboundedCache::new();
 /// 	let mut monitor_listener = (monitor, &*tx_broadcaster, &*fee_estimator, &*logger);
 /// 	let listeners = vec![
 /// 		(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();
+/// 	let (chain_cache, chain_tip) = init::synchronize_listeners(
+/// 		block_source, Network::Bitcoin, listeners).await.unwrap();
 ///
 /// 	// Allow the chain monitor to watch any channels.
 /// 	let monitor = monitor_listener.0;
@@ -130,7 +129,7 @@ where
 /// 	// Create an SPV client to notify the chain monitor and channel manager of block events.
 /// 	let chain_poller = poll::ChainPoller::new(block_source, Network::Bitcoin);
 /// 	let mut chain_listener = (chain_monitor, &manager);
-/// 	let spv_client = SpvClient::new(chain_tip, chain_poller, &mut cache, &chain_listener);
+/// 	let spv_client = SpvClient::new(chain_tip, chain_poller, chain_cache, &chain_listener);
 /// }
 /// ```
 ///
@@ -139,12 +138,11 @@ where
 /// [`ChannelMonitor`]: lightning::chain::channelmonitor::ChannelMonitor
 pub async fn synchronize_listeners<
 	B: Deref + Sized + Send + Sync,
-	C: Cache,
 	L: chain::Listen + ?Sized,
 >(
-	block_source: B, network: Network, header_cache: &mut C,
+	block_source: B, network: Network,
 	mut chain_listeners: Vec<(BestBlock, &L)>,
-) -> BlockSourceResult<ValidatedBlockHeader>
+) -> BlockSourceResult<(UnboundedCache, ValidatedBlockHeader)>
 where
 	B::Target: BlockSource,
 {
@@ -155,12 +153,12 @@ where
 	let mut chain_listeners_at_height = Vec::new();
 	let mut most_common_ancestor = None;
 	let mut most_connected_blocks = Vec::new();
+	let mut header_cache = UnboundedCache::new();
 	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 mut chain_notifier = ChainNotifier { header_cache: &mut header_cache, chain_listener };
 			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 {
@@ -180,32 +178,14 @@ where
 	// Connect new blocks for all listeners at once to avoid re-fetching blocks.
 	if let Some(common_ancestor) = most_common_ancestor {
 		let chain_listener = &ChainListenerSet(chain_listeners_at_height);
-		let mut chain_notifier = ChainNotifier { header_cache, chain_listener };
+		let mut chain_notifier = ChainNotifier { header_cache: &mut header_cache, chain_listener };
 		chain_notifier
 			.connect_blocks(common_ancestor, most_connected_blocks, &mut chain_poller)
 			.await
 			.map_err(|(e, _)| e)?;
 	}
 
-	Ok(best_header)
-}
-
-/// A wrapper to make a cache read-only.
-///
-/// Used to prevent losing headers that may be needed to disconnect blocks common to more than one
-/// listener.
-struct ReadOnlyCache<'a, C: Cache>(&'a mut C);
-
-impl<'a, C: Cache> Cache for ReadOnlyCache<'a, C> {
-	fn look_up(&self, block_hash: &BlockHash) -> Option<&ValidatedBlockHeader> {
-		self.0.look_up(block_hash)
-	}
-
-	fn block_connected(&mut self, _block_hash: BlockHash, _block_header: ValidatedBlockHeader) {
-		unreachable!()
-	}
-
-	fn blocks_disconnected(&mut self, _fork_point: &ValidatedBlockHeader) {}
+	Ok((header_cache, best_header))
 }
 
 /// Wrapper for supporting dynamically sized chain listeners.
@@ -273,9 +253,8 @@ mod tests {
 			(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 {
-			Ok(header) => assert_eq!(header, chain.tip()),
+		match synchronize_listeners(&chain, Network::Bitcoin, listeners).await {
+			Ok((_, header)) => assert_eq!(header, chain.tip()),
 			Err(e) => panic!("Unexpected error: {:?}", e),
 		}
 	}
@@ -305,11 +284,8 @@ mod tests {
 			(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));
-		cache.extend(fork_chain_3.header_cache(4..=4));
-		match synchronize_listeners(&main_chain, Network::Bitcoin, &mut cache, listeners).await {
-			Ok(header) => assert_eq!(header, main_chain.tip()),
+		match synchronize_listeners(&main_chain, Network::Bitcoin, listeners).await {
+			Ok((_, header)) => assert_eq!(header, main_chain.tip()),
 			Err(e) => panic!("Unexpected error: {:?}", e),
 		}
 	}
@@ -342,33 +318,8 @@ mod tests {
 			(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));
-		cache.extend(fork_chain_3.header_cache(4..=4));
-		match synchronize_listeners(&main_chain, Network::Bitcoin, &mut cache, listeners).await {
-			Ok(header) => assert_eq!(header, main_chain.tip()),
-			Err(e) => panic!("Unexpected error: {:?}", e),
-		}
-	}
-
-	#[tokio::test]
-	async fn cache_connected_and_keep_disconnected_blocks() {
-		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_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_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_best_block.block_hash));
-			},
+		match synchronize_listeners(&main_chain, Network::Bitcoin, listeners).await {
+			Ok((_, header)) => assert_eq!(header, main_chain.tip()),
 			Err(e) => panic!("Unexpected error: {:?}", e),
 		}
 	}

lightning-block-sync/src/lib.rs

@@ -170,18 +170,13 @@ pub enum BlockData {
 /// sources for the best chain tip. During this process it detects any chain forks, determines which
 /// constitutes the best chain, and updates the listener accordingly with any blocks that were
 /// connected or disconnected since the last poll.
-///
-/// Block headers for the best chain are maintained in the parameterized cache, allowing for a
-/// custom cache eviction policy. This offers flexibility to those sensitive to resource usage.
-/// Hence, there is a trade-off between a lower memory footprint and potentially increased network
-/// I/O as headers are re-fetched during fork detection.
-pub struct SpvClient<'a, P: Poll, C: Cache, L: Deref>
+pub struct SpvClient<P: Poll, L: Deref>
 where
 	L::Target: chain::Listen,
 {
 	chain_tip: ValidatedBlockHeader,
 	chain_poller: P,
-	chain_notifier: ChainNotifier<'a, C, L>,
+	chain_notifier: ChainNotifier<UnboundedCache, L>,
 }
 
 /// The `Cache` trait defines behavior for managing a block header cache, where block headers are
@@ -194,7 +189,7 @@ where
 /// Implementations may define how long to retain headers such that it's unlikely they will ever be
 /// needed to disconnect a block.  In cases where block sources provide access to headers on stale
 /// forks reliably, caches may be entirely unnecessary.
-pub trait Cache {
+pub(crate) trait Cache {
 	/// Retrieves the block header keyed by the given block hash.
 	fn look_up(&self, block_hash: &BlockHash) -> Option<&ValidatedBlockHeader>;
 
@@ -226,7 +221,21 @@ impl Cache for UnboundedCache {
 	}
 }
 
-impl<'a, P: Poll, C: Cache, L: Deref> SpvClient<'a, P, C, L>
+impl Cache for &mut UnboundedCache {
+	fn look_up(&self, block_hash: &BlockHash) -> Option<&ValidatedBlockHeader> {
+		self.get(block_hash)
+	}
+
+	fn block_connected(&mut self, block_hash: BlockHash, block_header: ValidatedBlockHeader) {
+		self.insert(block_hash, block_header);
+	}
+
+	fn blocks_disconnected(&mut self, fork_point: &ValidatedBlockHeader) {
+		self.retain(|_, block_info| block_info.height < fork_point.height);
+	}
+}
+
+impl<P: Poll, L: Deref> SpvClient<P, L>
 where
 	L::Target: chain::Listen,
 {
@@ -241,7 +250,7 @@ where
 	///
 	/// [`poll_best_tip`]: SpvClient::poll_best_tip
 	pub fn new(
-		chain_tip: ValidatedBlockHeader, chain_poller: P, header_cache: &'a mut C,
+		chain_tip: ValidatedBlockHeader, chain_poller: P, header_cache: UnboundedCache,
 		chain_listener: L,
 	) -> Self {
 		let chain_notifier = ChainNotifier { header_cache, chain_listener };
@@ -295,15 +304,15 @@ where
 /// Notifies [listeners] of blocks that have been connected or disconnected from the chain.
 ///
 /// [listeners]: lightning::chain::Listen
-pub struct ChainNotifier<'a, C: Cache, L: Deref>
+pub(crate) struct ChainNotifier<C: Cache, L: Deref>
 where
 	L::Target: chain::Listen,
 {
 	/// Cache for looking up headers before fetching from a block source.
-	header_cache: &'a mut C,
+	pub(crate) header_cache: C,
 
 	/// Listener that will be notified of connected or disconnected blocks.
-	chain_listener: L,
+	pub(crate) chain_listener: L,
 }
 
 /// Changes made to the chain between subsequent polls that transformed it from having one chain tip
@@ -321,7 +330,7 @@ struct ChainDifference {
 	connected_blocks: Vec<ValidatedBlockHeader>,
 }
 
-impl<'a, C: Cache, L: Deref> ChainNotifier<'a, C, L>
+impl<C: Cache, L: Deref> ChainNotifier<C, L>
 where
 	L::Target: chain::Listen,
 {
@@ -480,9 +489,9 @@ mod spv_client_tests {
 		let best_tip = chain.at_height(1);
 
 		let poller = poll::ChainPoller::new(&mut chain, Network::Testnet);
-		let mut cache = UnboundedCache::new();
+		let cache = UnboundedCache::new();
 		let mut listener = NullChainListener {};
-		let mut client = SpvClient::new(best_tip, poller, &mut cache, &mut listener);
+		let mut client = SpvClient::new(best_tip, poller, cache, &mut listener);
 		match client.poll_best_tip().await {
 			Err(e) => {
 				assert_eq!(e.kind(), BlockSourceErrorKind::Persistent);
@@ -499,9 +508,9 @@ mod spv_client_tests {
 		let common_tip = chain.tip();
 
 		let poller = poll::ChainPoller::new(&mut chain, Network::Testnet);
-		let mut cache = UnboundedCache::new();
+		let cache = UnboundedCache::new();
 		let mut listener = NullChainListener {};
-		let mut client = SpvClient::new(common_tip, poller, &mut cache, &mut listener);
+		let mut client = SpvClient::new(common_tip, poller, cache, &mut listener);
 		match client.poll_best_tip().await {
 			Err(e) => panic!("Unexpected error: {:?}", e),
 			Ok((chain_tip, blocks_connected)) => {
@@ -519,9 +528,9 @@ mod spv_client_tests {
 		let old_tip = chain.at_height(1);
 
 		let poller = poll::ChainPoller::new(&mut chain, Network::Testnet);
-		let mut cache = UnboundedCache::new();
+		let cache = UnboundedCache::new();
 		let mut listener = NullChainListener {};
-		let mut client = SpvClient::new(old_tip, poller, &mut cache, &mut listener);
+		let mut client = SpvClient::new(old_tip, poller, cache, &mut listener);
 		match client.poll_best_tip().await {
 			Err(e) => panic!("Unexpected error: {:?}", e),
 			Ok((chain_tip, blocks_connected)) => {
@@ -539,9 +548,9 @@ mod spv_client_tests {
 		let old_tip = chain.at_height(1);
 
 		let poller = poll::ChainPoller::new(&mut chain, Network::Testnet);
-		let mut cache = UnboundedCache::new();
+		let cache = UnboundedCache::new();
 		let mut listener = NullChainListener {};
-		let mut client = SpvClient::new(old_tip, poller, &mut cache, &mut listener);
+		let mut client = SpvClient::new(old_tip, poller, cache, &mut listener);
 		match client.poll_best_tip().await {
 			Err(e) => panic!("Unexpected error: {:?}", e),
 			Ok((chain_tip, blocks_connected)) => {
@@ -559,9 +568,9 @@ mod spv_client_tests {
 		let old_tip = chain.at_height(1);
 
 		let poller = poll::ChainPoller::new(&mut chain, Network::Testnet);
-		let mut cache = UnboundedCache::new();
+		let cache = UnboundedCache::new();
 		let mut listener = NullChainListener {};
-		let mut client = SpvClient::new(old_tip, poller, &mut cache, &mut listener);
+		let mut client = SpvClient::new(old_tip, poller, cache, &mut listener);
 		match client.poll_best_tip().await {
 			Err(e) => panic!("Unexpected error: {:?}", e),
 			Ok((chain_tip, blocks_connected)) => {
@@ -580,9 +589,9 @@ mod spv_client_tests {
 		let worse_tip = chain.tip();
 
 		let poller = poll::ChainPoller::new(&mut chain, Network::Testnet);
-		let mut cache = UnboundedCache::new();
+		let cache = UnboundedCache::new();
 		let mut listener = NullChainListener {};
-		let mut client = SpvClient::new(best_tip, poller, &mut cache, &mut listener);
+		let mut client = SpvClient::new(best_tip, poller, cache, &mut listener);
 		match client.poll_best_tip().await {
 			Err(e) => panic!("Unexpected error: {:?}", e),
 			Ok((chain_tip, blocks_connected)) => {