Make UnboundedCache bounded

In the previous commit we moved to hard-coding `UnboundedCache` in
the `lightning-block-sync` interface. This is great, except that
its an unbounded cache that can use arbitrary amounts of memory
(though never really all that much - its just headers that come in
while we're running).

Here we simply limit the size, and while we're at it give it a more
generic `HeaderCache` name.

Author: TheBlueMatt

lightning-block-sync/src/init.rs

@@ -2,10 +2,9 @@
 //! from disk.
 
 use crate::poll::{ChainPoller, Validate, ValidatedBlockHeader};
-use crate::{BlockSource, BlockSourceResult, Cache, ChainNotifier, UnboundedCache};
+use crate::{BlockSource, BlockSourceResult, ChainNotifier, HeaderCache};
 
 use bitcoin::block::Header;
-use bitcoin::hash_types::BlockHash;
 use bitcoin::network::Network;
 
 use lightning::chain;
@@ -138,7 +137,7 @@ where
 /// [`ChannelMonitor`]: lightning::chain::channelmonitor::ChannelMonitor
 pub async fn synchronize_listeners<B: Deref + Sized + Send + Sync, L: chain::Listen + ?Sized>(
 	block_source: B, network: Network, mut chain_listeners: Vec<(BestBlock, &L)>,
-) -> BlockSourceResult<(UnboundedCache, ValidatedBlockHeader)>
+) -> BlockSourceResult<(HeaderCache, ValidatedBlockHeader)>
 where
 	B::Target: BlockSource,
 {
@@ -149,7 +148,7 @@ 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();
+	let mut header_cache = HeaderCache::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 (common_ancestor, connected_blocks) = {

lightning-block-sync/src/lib.rs

@@ -176,7 +176,7 @@ where
 {
 	chain_tip: ValidatedBlockHeader,
 	chain_poller: P,
-	chain_notifier: ChainNotifier<UnboundedCache, L>,
+	chain_notifier: ChainNotifier<HeaderCache, L>,
 }
 
 /// The `Cache` trait defines behavior for managing a block header cache, where block headers are
@@ -204,34 +204,47 @@ pub(crate) trait Cache {
 	fn blocks_disconnected(&mut self, fork_point: &ValidatedBlockHeader);
 }
 
-/// Unbounded cache of block headers keyed by block hash.
-pub type UnboundedCache = std::collections::HashMap<BlockHash, ValidatedBlockHeader>;
+/// Bounded cache of block headers keyed by block hash.
+///
+/// Retains only the last `ANTI_REORG_DELAY * 2` block headers based on height.
+pub struct HeaderCache(std::collections::HashMap<BlockHash, ValidatedBlockHeader>);
 
-impl Cache for UnboundedCache {
+impl HeaderCache {
+	/// Creates a new empty header cache.
+	pub fn new() -> Self {
+		Self(std::collections::HashMap::new())
+	}
+}
+
+impl Cache for HeaderCache {
 	fn look_up(&self, block_hash: &BlockHash) -> Option<&ValidatedBlockHeader> {
-		self.get(block_hash)
+		self.0.get(block_hash)
 	}
 
 	fn block_connected(&mut self, block_hash: BlockHash, block_header: ValidatedBlockHeader) {
-		self.insert(block_hash, block_header);
+		self.0.insert(block_hash, block_header);
+
+		// Remove headers older than a week.
+		let cutoff_height = block_header.height.saturating_sub(6 * 24 * 7);
+		self.0.retain(|_, header| header.height >= cutoff_height);
 	}
 
 	fn blocks_disconnected(&mut self, fork_point: &ValidatedBlockHeader) {
-		self.retain(|_, block_info| block_info.height < fork_point.height);
+		self.0.retain(|_, block_info| block_info.height < fork_point.height);
 	}
 }
 
-impl Cache for &mut UnboundedCache {
+impl Cache for &mut HeaderCache {
 	fn look_up(&self, block_hash: &BlockHash) -> Option<&ValidatedBlockHeader> {
-		self.get(block_hash)
+		self.0.get(block_hash)
 	}
 
 	fn block_connected(&mut self, block_hash: BlockHash, block_header: ValidatedBlockHeader) {
-		self.insert(block_hash, block_header);
+		(*self).block_connected(block_hash, block_header);
 	}
 
 	fn blocks_disconnected(&mut self, fork_point: &ValidatedBlockHeader) {
-		self.retain(|_, block_info| block_info.height < fork_point.height);
+		self.0.retain(|_, block_info| block_info.height < fork_point.height);
 	}
 }
 
@@ -250,7 +263,7 @@ where
 	///
 	/// [`poll_best_tip`]: SpvClient::poll_best_tip
 	pub fn new(
-		chain_tip: ValidatedBlockHeader, chain_poller: P, header_cache: UnboundedCache,
+		chain_tip: ValidatedBlockHeader, chain_poller: P, header_cache: HeaderCache,
 		chain_listener: L,
 	) -> Self {
 		let chain_notifier = ChainNotifier { header_cache, chain_listener };
@@ -490,7 +503,7 @@ mod spv_client_tests {
 		let best_tip = chain.at_height(1);
 
 		let poller = poll::ChainPoller::new(&mut chain, Network::Testnet);
-		let cache = UnboundedCache::new();
+		let cache = HeaderCache::new();
 		let mut listener = NullChainListener {};
 		let mut client = SpvClient::new(best_tip, poller, cache, &mut listener);
 		match client.poll_best_tip().await {
@@ -509,7 +522,7 @@ mod spv_client_tests {
 		let common_tip = chain.tip();
 
 		let poller = poll::ChainPoller::new(&mut chain, Network::Testnet);
-		let cache = UnboundedCache::new();
+		let cache = HeaderCache::new();
 		let mut listener = NullChainListener {};
 		let mut client = SpvClient::new(common_tip, poller, cache, &mut listener);
 		match client.poll_best_tip().await {
@@ -529,7 +542,7 @@ mod spv_client_tests {
 		let old_tip = chain.at_height(1);
 
 		let poller = poll::ChainPoller::new(&mut chain, Network::Testnet);
-		let cache = UnboundedCache::new();
+		let cache = HeaderCache::new();
 		let mut listener = NullChainListener {};
 		let mut client = SpvClient::new(old_tip, poller, cache, &mut listener);
 		match client.poll_best_tip().await {
@@ -549,7 +562,7 @@ mod spv_client_tests {
 		let old_tip = chain.at_height(1);
 
 		let poller = poll::ChainPoller::new(&mut chain, Network::Testnet);
-		let cache = UnboundedCache::new();
+		let cache = HeaderCache::new();
 		let mut listener = NullChainListener {};
 		let mut client = SpvClient::new(old_tip, poller, cache, &mut listener);
 		match client.poll_best_tip().await {
@@ -569,7 +582,7 @@ mod spv_client_tests {
 		let old_tip = chain.at_height(1);
 
 		let poller = poll::ChainPoller::new(&mut chain, Network::Testnet);
-		let cache = UnboundedCache::new();
+		let cache = HeaderCache::new();
 		let mut listener = NullChainListener {};
 		let mut client = SpvClient::new(old_tip, poller, cache, &mut listener);
 		match client.poll_best_tip().await {
@@ -590,7 +603,7 @@ mod spv_client_tests {
 		let worse_tip = chain.tip();
 
 		let poller = poll::ChainPoller::new(&mut chain, Network::Testnet);
-		let cache = UnboundedCache::new();
+		let cache = HeaderCache::new();
 		let mut listener = NullChainListener {};
 		let mut client = SpvClient::new(best_tip, poller, cache, &mut listener);
 		match client.poll_best_tip().await {

lightning-block-sync/src/test_utils.rs

@@ -1,6 +1,7 @@
 use crate::poll::{Validate, ValidatedBlockHeader};
 use crate::{
-	BlockData, BlockHeaderData, BlockSource, BlockSourceError, BlockSourceResult, UnboundedCache,
+	BlockData, BlockHeaderData, BlockSource, BlockSourceError, BlockSourceResult, Cache,
+	HeaderCache,
 };
 
 use bitcoin::block::{Block, Header, Version};
@@ -144,12 +145,12 @@ impl Blockchain {
 		self.blocks.pop()
 	}
 
-	pub fn header_cache(&self, heights: std::ops::RangeInclusive<usize>) -> UnboundedCache {
-		let mut cache = UnboundedCache::new();
+	pub fn header_cache(&self, heights: std::ops::RangeInclusive<usize>) -> HeaderCache {
+		let mut cache = HeaderCache::new();
 		for i in heights {
 			let value = self.at_height(i);
 			let key = value.header.block_hash();
-			assert!(cache.insert(key, value).is_none());
+			cache.block_connected(key, value);
 		}
 		cache
 	}