Implement ChainMultiplexer in terms of ChainPoller

Removes the BlockSource implementation for ChainMultiplexer as it is no
longer needed.

Author: jkczyz

lightning-block-sync/src/poller.rs

@@ -1,7 +1,5 @@
-use crate::{AsyncBlockSourceResult, BlockHeaderData, BlockSource, BlockSourceError, ChainTip, Poll, Validate, ValidatedBlock, ValidatedBlockHeader};
+use crate::{AsyncBlockSourceResult, BlockSource, BlockSourceError, ChainTip, Poll, Validate, ValidatedBlock, ValidatedBlockHeader};
 
-use bitcoin::blockdata::block::Block;
-use bitcoin::hash_types::BlockHash;
 use bitcoin::network::constants::Network;
 
 use std::ops::DerefMut;
@@ -73,27 +71,26 @@ impl<'b, B: DerefMut<Target=dyn BlockSource + 'b> + Sized + Sync + Send> Poll fo
 }
 
 pub struct ChainMultiplexer<'b, B: DerefMut<Target=dyn BlockSource + 'b> + Sized + Sync + Send> {
-	block_sources: Vec<(B, BlockSourceError)>,
-	backup_block_sources: Vec<(B, BlockSourceError)>,
+	block_sources: Vec<(ChainPoller<'b, B>, BlockSourceError)>,
+	backup_block_sources: Vec<(ChainPoller<'b, B>, BlockSourceError)>,
 	best_block_source: usize,
-	network: Network,
 }
 
 impl<'b, B: DerefMut<Target=dyn BlockSource + 'b> + Sized + Sync + Send> ChainMultiplexer<'b, B> {
 	pub fn new(mut block_sources: Vec<B>, mut backup_block_sources: Vec<B>, network: Network) -> Self {
 		assert!(!block_sources.is_empty());
 		let block_sources = block_sources.drain(..).map(|block_source| {
-			(block_source, BlockSourceError::Transient)
+			(ChainPoller::new(block_source, network), BlockSourceError::Transient)
 		}).collect();
 
 		let backup_block_sources = backup_block_sources.drain(..).map(|block_source| {
-			(block_source, BlockSourceError::Transient)
+			(ChainPoller::new(block_source, network), BlockSourceError::Transient)
 		}).collect();
 
-		Self { block_sources, backup_block_sources, best_block_source: 0, network }
+		Self { block_sources, backup_block_sources, best_block_source: 0 }
 	}
 
-	fn best_and_backup_block_sources(&mut self) -> Vec<&mut (B, BlockSourceError)> {
+	fn best_and_backup_block_sources(&mut self) -> Vec<&mut (ChainPoller<'b, B>, BlockSourceError)> {
 		let best_block_source = self.block_sources.get_mut(self.best_block_source).unwrap();
 		let backup_block_sources = self.backup_block_sources.iter_mut();
 		std::iter::once(best_block_source)
@@ -110,12 +107,11 @@ impl<'b, B: 'b + DerefMut<Target=dyn BlockSource + 'b> + Sized + Sync + Send> Po
 		Box::pin(async move {
 			let mut heaviest_chain_tip = best_chain_tip;
 			let mut best_result = Err(BlockSourceError::Persistent);
-			for (i, (block_source, error)) in self.block_sources.iter_mut().enumerate() {
+			for (i, (poller, error)) in self.block_sources.iter_mut().enumerate() {
 				if let BlockSourceError::Persistent = error {
 					continue;
 				}
 
-				let mut poller = ChainPoller::new(&mut **block_source as &mut dyn BlockSource, self.network);
 				let result = poller.poll_chain_tip(heaviest_chain_tip).await;
 				match result {
 					Err(BlockSourceError::Persistent) => {
@@ -152,41 +148,8 @@ impl<'b, B: 'b + DerefMut<Target=dyn BlockSource + 'b> + Sized + Sync + Send> Po
 		AsyncBlockSourceResult<'a, ValidatedBlockHeader>
 	{
 		Box::pin(async move {
-			let previous_hash = &header.header.prev_blockhash;
-			let height = header.height - 1;
-			self.get_header(previous_hash, Some(height)).await?
-				.validate(*previous_hash)
-		})
-	}
-
-	fn fetch_block<'a>(&'a mut self, header: &'a ValidatedBlockHeader) ->
-		AsyncBlockSourceResult<'a, ValidatedBlock>
-	{
-		Box::pin(async move {
-			self.get_block(&header.block_hash).await?
-				.validate(header.block_hash)
-		})
-	}
-}
-
-impl<'b, B: DerefMut<Target=dyn BlockSource + 'b> + Sized + Sync + Send> BlockSource for ChainMultiplexer<'b, B> {
-	fn get_header<'a>(&'a mut self, header_hash: &'a BlockHash, height: Option<u32>) -> AsyncBlockSourceResult<'a, BlockHeaderData> {
-		Box::pin(async move {
-			for (block_source, error) in self.best_and_backup_block_sources() {
-				let result = block_source.get_header(header_hash, height).await;
-				match result {
-					Err(e) => *error = e,
-					Ok(_) => return result,
-				}
-			}
-			Err(BlockSourceError::Persistent)
-		})
-	}
-
-	fn get_block<'a>(&'a mut self, header_hash: &'a BlockHash) -> AsyncBlockSourceResult<'a, Block> {
-		Box::pin(async move {
-			for (block_source, error) in self.best_and_backup_block_sources() {
-				let result = block_source.get_block(header_hash).await;
+			for (poller, error) in self.best_and_backup_block_sources() {
+				let result = poller.look_up_previous_header(header).await;
 				match result {
 					Err(e) => *error = e,
 					Ok(_) => return result,
@@ -196,10 +159,12 @@ impl<'b, B: DerefMut<Target=dyn BlockSource + 'b> + Sized + Sync + Send> BlockSo
 		})
 	}
 
-	fn get_best_block<'a>(&'a mut self) -> AsyncBlockSourceResult<'a, (BlockHash, Option<u32>)> {
+	fn fetch_block<'a>(&'a mut self, header: &'a ValidatedBlockHeader) ->
+		AsyncBlockSourceResult<'a, ValidatedBlock>
+	{
 		Box::pin(async move {
-			for (block_source, error) in self.best_and_backup_block_sources() {
-				let result = block_source.get_best_block().await;
+			for (poller, error) in self.best_and_backup_block_sources() {
+				let result = poller.fetch_block(header).await;
 				match result {
 					Err(e) => *error = e,
 					Ok(_) => return result,