lightning-block-sync 0.3.0-beta1

//! A lightweight client for keeping in sync with chain activity.
//!
//! Defines an [`SpvClient`] utility for polling one or more block sources for the best chain tip.
//! It is used to notify listeners of blocks connected or disconnected since the last poll. Useful
//! for keeping a Lightning node in sync with the chain.
//!
//! Defines a [`BlockSource`] trait, which is an asynchronous interface for retrieving block headers
//! and data.
//!
//! Enabling feature `rest-client` or `rpc-client` allows configuring the client to fetch blocks
//! using Bitcoin Core's REST or RPC interface, respectively.
//!
//! Both features support either blocking I/O using `std::net::TcpStream` or, with feature `tokio`,
//! non-blocking I/O using `tokio::net::TcpStream` from inside a Tokio runtime.

#![deny(rustdoc::broken_intra_doc_links)]
#![deny(rustdoc::private_intra_doc_links)]
#![deny(missing_docs)]
#![cfg_attr(docsrs, feature(doc_cfg))]

extern crate alloc;
extern crate core;

#[cfg(any(feature = "rest-client", feature = "rpc-client"))]
pub mod http;

pub mod init;
pub mod poll;

pub mod gossip;

#[cfg(feature = "rest-client")]
pub mod rest;

#[cfg(feature = "rpc-client")]
pub mod rpc;

#[cfg(any(feature = "rest-client", feature = "rpc-client"))]
mod convert;

#[cfg(test)]
mod test_utils;

#[cfg(any(feature = "rest-client", feature = "rpc-client"))]
mod utils;

#[allow(unused)]
mod async_poll;

use crate::poll::{ChainTip, Poll, ValidatedBlockHeader};

use bitcoin::block::{Block, Header};
use bitcoin::hash_types::BlockHash;
use bitcoin::pow::Work;

use lightning::chain;
use lightning::chain::BlockLocator;

use std::future::Future;
use std::ops::Deref;

/// Abstract type for retrieving block headers and data.
pub trait BlockSource: Sync + Send {
	/// Returns the header for a given hash. A height hint may be provided in case a block source
	/// cannot easily find headers based on a hash. This is merely a hint and thus the returned
	/// header must have the same hash as was requested. Otherwise, an error must be returned.
	///
	/// Implementations that cannot find headers based on the hash should return a `Transient` error
	/// when `height_hint` is `None`.
	fn get_header<'a>(
		&'a self, header_hash: &'a BlockHash, height_hint: Option<u32>,
	) -> impl Future<Output = BlockSourceResult<BlockHeaderData>> + Send + 'a;

	/// Returns the block for a given hash. A headers-only block source should return a `Transient`
	/// error.
	fn get_block<'a>(
		&'a self, header_hash: &'a BlockHash,
	) -> impl Future<Output = BlockSourceResult<BlockData>> + Send + 'a;

	/// Returns the hash of the best block and, optionally, its height.
	///
	/// When polling a block source, [`Poll`] implementations may pass the height to [`get_header`]
	/// to allow for a more efficient lookup.
	///
	/// [`get_header`]: Self::get_header
	fn get_best_block<'a>(
		&'a self,
	) -> impl Future<Output = BlockSourceResult<(BlockHash, Option<u32>)>> + Send + 'a;
}

/// Result type for `BlockSource` requests.
pub type BlockSourceResult<T> = Result<T, BlockSourceError>;

/// Error type for `BlockSource` requests.
///
/// Transient errors may be resolved when re-polling, but no attempt will be made to re-poll on
/// persistent errors.
#[derive(Debug)]
pub struct BlockSourceError {
	kind: BlockSourceErrorKind,
	error: Box<dyn std::error::Error + Send + Sync>,
}

/// The kind of `BlockSourceError`, either persistent or transient.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum BlockSourceErrorKind {
	/// Indicates an error that won't resolve when retrying a request (e.g., invalid data).
	Persistent,

	/// Indicates an error that may resolve when retrying a request (e.g., unresponsive).
	Transient,
}

impl BlockSourceError {
	/// Creates a new persistent error originated from the given error.
	pub fn persistent<E>(error: E) -> Self
	where
		E: Into<Box<dyn std::error::Error + Send + Sync>>,
	{
		Self { kind: BlockSourceErrorKind::Persistent, error: error.into() }
	}

	/// Creates a new transient error originated from the given error.
	pub fn transient<E>(error: E) -> Self
	where
		E: Into<Box<dyn std::error::Error + Send + Sync>>,
	{
		Self { kind: BlockSourceErrorKind::Transient, error: error.into() }
	}

	/// Returns the kind of error.
	pub fn kind(&self) -> BlockSourceErrorKind {
		self.kind
	}

	/// Converts the error into the underlying error.
	///
	/// May contain an [`std::io::Error`] from the [`BlockSource`]. See implementations for further
	/// details, if any.
	pub fn into_inner(self) -> Box<dyn std::error::Error + Send + Sync> {
		self.error
	}
}

/// A block header and some associated data. This information should be available from most block
/// sources (and, notably, is available in Bitcoin Core's RPC and REST interfaces).
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct BlockHeaderData {
	/// The block header itself.
	pub header: Header,

	/// The block height where the genesis block has height 0.
	pub height: u32,

	/// The total chain work required to build a chain of equivalent weight.
	pub chainwork: Work,
}

/// A block including either all its transactions or only the block header.
///
/// [`BlockSource`] may be implemented to either always return full blocks or, in the case of
/// compact block filters (BIP 157/158), return header-only blocks when no pertinent transactions
/// match. See [`chain::Filter`] for details on how to notify a source of such transactions.
pub enum BlockData {
	/// A block containing all its transactions.
	FullBlock(Block),
	/// A block header for when the block does not contain any pertinent transactions.
	HeaderOnly(Header),
}

/// A lightweight client for keeping a listener in sync with the chain, allowing for Simplified
/// Payment Verification (SPV).
///
/// The client is parameterized by a chain poller which is responsible for polling one or more block
/// 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.
pub struct SpvClient<P: Poll, L: Deref>
where
	L::Target: chain::Listen,
{
	chain_tip: ValidatedBlockHeader,
	chain_poller: P,
	header_cache: HeaderCache,
	chain_listener: L,
}

/// The maximum number of [`ValidatedBlockHeader`]s stored in a [`HeaderCache`].
pub const HEADER_CACHE_LIMIT: u32 = 6 * 24 * 7;

/// Bounded cache of block headers keyed by block hash.
///
/// 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(), retain_on_disconnect: false }
	}

	/// Retrieves the block header keyed by the given block hash.
	pub fn look_up(&self, block_hash: &BlockHash) -> Option<&ValidatedBlockHeader> {
		self.headers.get(block_hash)
	}

	/// Called when a block has been connected to the best chain to ensure it is available to be
	/// disconnected later if needed.
	pub(crate) fn block_connected(
		&mut self, block_hash: BlockHash, block_header: ValidatedBlockHeader,
	) {
		self.headers.insert(block_hash, block_header);

		// Remove headers older than a week.
		let cutoff_height = block_header.height.saturating_sub(HEADER_CACHE_LIMIT);
		self.headers.retain(|_, header| header.height >= cutoff_height);
	}

	/// Inserts the given block header during a find_difference operation, implying it might not be
	/// the best header.
	pub(crate) fn insert_during_diff(
		&mut self, block_hash: BlockHash, block_header: ValidatedBlockHeader,
	) {
		self.headers.insert(block_hash, block_header);

		// Remove headers older than our newest header minus a week.
		let best_height = self.headers.iter().map(|(_, header)| header.height).max().unwrap_or(0);
		let cutoff_height = best_height.saturating_sub(HEADER_CACHE_LIMIT);
		self.headers.retain(|_, header| header.height >= cutoff_height);
	}

	/// Called when blocks have been disconnected from the best chain. Only the fork point
	/// (best common ancestor) is provided.
	///
	/// 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) {
		if !self.retain_on_disconnect {
			self.headers.retain(|_, block_info| block_info.height <= fork_point.height);
		}
	}
}

impl<P: Poll, L: Deref> SpvClient<P, L>
where
	L::Target: chain::Listen,
{
	/// Creates a new SPV client using `chain_tip` as the best known chain tip.
	///
	/// Subsequent calls to [`poll_best_tip`] will poll for the best chain tip using the given chain
	/// poller, which may be configured with one or more block sources to query. At least one block
	/// source must provide headers back from the best chain tip to its common ancestor with
	/// `chain_tip`.
	/// * `header_cache` is used to look up and store headers on the best chain
	/// * `chain_listener` is notified of any blocks connected or disconnected
	///
	/// [`poll_best_tip`]: SpvClient::poll_best_tip
	pub fn new(
		chain_tip: ValidatedBlockHeader, chain_poller: P, header_cache: HeaderCache,
		chain_listener: L,
	) -> Self {
		Self { chain_tip, chain_poller, header_cache, chain_listener }
	}

	/// Polls for the best tip and updates the chain listener with any connected or disconnected
	/// blocks accordingly.
	///
	/// Returns the best polled chain tip relative to the previous best known tip and whether any
	/// blocks were indeed connected or disconnected.
	pub async fn poll_best_tip(&mut self) -> BlockSourceResult<(ChainTip, bool)> {
		let chain_tip = self.chain_poller.poll_chain_tip(self.chain_tip).await?;
		let blocks_connected = match chain_tip {
			ChainTip::Common => false,
			ChainTip::Better(chain_tip) => {
				debug_assert_ne!(chain_tip.block_hash, self.chain_tip.block_hash);
				debug_assert!(chain_tip.chainwork > self.chain_tip.chainwork);
				self.update_chain_tip(chain_tip).await
			},
			ChainTip::Worse(chain_tip) => {
				debug_assert_ne!(chain_tip.block_hash, self.chain_tip.block_hash);
				debug_assert!(chain_tip.chainwork <= self.chain_tip.chainwork);
				false
			},
		};
		Ok((chain_tip, blocks_connected))
	}

	/// Updates the chain tip, syncing the chain listener with any connected or disconnected
	/// blocks. Returns whether there were any such blocks.
	async fn update_chain_tip(&mut self, best_chain_tip: ValidatedBlockHeader) -> bool {
		let mut chain_notifier = ChainNotifier {
			header_cache: &mut self.header_cache,
			chain_listener: &*self.chain_listener,
		};
		match chain_notifier
			.synchronize_listener(best_chain_tip, &self.chain_tip, &mut self.chain_poller)
			.await
		{
			Ok(_) => {
				self.chain_tip = best_chain_tip;
				true
			},
			Err((_, Some(chain_tip))) if chain_tip.block_hash != self.chain_tip.block_hash => {
				self.chain_tip = chain_tip;
				true
			},
			Err(_) => false,
		}
	}
}

/// Notifies [listeners] of blocks that have been connected or disconnected from the chain.
///
/// [listeners]: lightning::chain::Listen
pub(crate) struct ChainNotifier<'a, L: chain::Listen + ?Sized> {
	/// Cache for looking up headers before fetching from a block source.
	pub(crate) header_cache: &'a mut HeaderCache,

	/// Listener that will be notified of connected or disconnected blocks.
	pub(crate) chain_listener: &'a L,
}

/// Changes made to the chain between subsequent polls that transformed it from having one chain tip
/// to another.
///
/// Blocks are given in height-descending order. Therefore, blocks are first disconnected in order
/// before new blocks are connected in reverse order.
struct ChainDifference {
	/// The most recent ancestor common between the chain tips.
	///
	/// If there are any disconnected blocks, this is where the chain forked.
	common_ancestor: ValidatedBlockHeader,

	/// Blocks that were connected to the chain since the last poll.
	connected_blocks: Vec<ValidatedBlockHeader>,
}

impl<'a, L: chain::Listen + ?Sized> ChainNotifier<'a, L> {
	/// Finds the first common ancestor between `new_header` and `old_header`, disconnecting blocks
	/// from `old_header` to get to that point and then connecting blocks until `new_header`.
	///
	/// Validates headers along the transition path, but doesn't fetch blocks until the chain is
	/// disconnected to the fork point. Thus, this may return an `Err` that includes where the tip
	/// ended up which may not be `new_header`. Note that the returned `Err` contains `Some` header
	/// if and only if the transition from `old_header` to `new_header` is valid.
	async fn synchronize_listener<P: Poll>(
		&mut self, new_header: ValidatedBlockHeader, old_header: &ValidatedBlockHeader,
		chain_poller: &mut P,
	) -> Result<(), (BlockSourceError, Option<ValidatedBlockHeader>)> {
		let difference = self
			.find_difference_from_header(new_header, old_header, chain_poller)
			.await
			.map_err(|e| (e, None))?;
		if difference.common_ancestor != *old_header {
			self.disconnect_blocks(difference.common_ancestor);
		}
		self.connect_blocks(difference.common_ancestor, difference.connected_blocks, chain_poller)
			.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.
	///
	/// Updates the header cache as it goes, tracking headers needed to find the diff to reuse for
	/// other objects that might need similar headers.
	async fn find_difference_from_best_block<P: Poll>(
		&mut self, current_header: ValidatedBlockHeader, prev_best_block: BlockLocator,
		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(
					"BlockLocator had more previous_blocks than its height",
				),
			)?;
			if let Ok(header) = chain_poller.get_header(block_hash, Some(height)).await {
				found_header = Some(header);
				self.header_cache.insert_during_diff(*block_hash, header);
				break;
			}
		}
		let found_header = found_header.ok_or_else(|| {
			BlockSourceError::persistent("could not resolve any block from BlockLocator")
		})?;

		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_from_header<P: Poll>(
		&self, current_header: ValidatedBlockHeader, prev_header: &ValidatedBlockHeader,
		chain_poller: &mut P,
	) -> BlockSourceResult<ChainDifference> {
		let mut connected_blocks = Vec::new();
		let mut current = current_header;
		let mut previous = *prev_header;
		loop {
			// Found the common ancestor.
			if current.block_hash == previous.block_hash {
				break;
			}

			// Walk back the chain, finding blocks needed to connect and disconnect. Only walk back
			// the header with the greater height, or both if equal heights.
			let current_height = current.height;
			let previous_height = previous.height;
			if current_height <= previous_height {
				previous = self.look_up_previous_header(chain_poller, &previous).await?;
			}
			if current_height >= previous_height {
				connected_blocks.push(current);
				current = self.look_up_previous_header(chain_poller, &current).await?;
			}
		}

		let common_ancestor = current;
		Ok(ChainDifference { common_ancestor, connected_blocks })
	}

	/// Returns the previous header for the given header, either by looking it up in the cache or
	/// fetching it if not found.
	async fn look_up_previous_header<P: Poll>(
		&self, chain_poller: &mut P, header: &ValidatedBlockHeader,
	) -> BlockSourceResult<ValidatedBlockHeader> {
		match self.header_cache.look_up(&header.header.prev_blockhash) {
			Some(prev_header) => Ok(*prev_header),
			None => chain_poller.look_up_previous_header(header).await,
		}
	}

	/// Notifies the chain listeners of disconnected blocks.
	fn disconnect_blocks(&mut self, fork_point: ValidatedBlockHeader) {
		self.header_cache.blocks_disconnected(&fork_point);
		let best_block = BlockLocator::new(fork_point.block_hash, fork_point.height);
		self.chain_listener.blocks_disconnected(best_block);
	}

	/// Notifies the chain listeners of connected blocks.
	async fn connect_blocks<P: Poll>(
		&mut self, mut new_tip: ValidatedBlockHeader,
		mut connected_blocks: Vec<ValidatedBlockHeader>, chain_poller: &mut P,
	) -> Result<(), (BlockSourceError, Option<ValidatedBlockHeader>)> {
		for header in connected_blocks.drain(..).rev() {
			let height = header.height;
			let block_data =
				chain_poller.fetch_block(&header).await.map_err(|e| (e, Some(new_tip)))?;
			debug_assert_eq!(block_data.block_hash, header.block_hash);

			match block_data.deref() {
				BlockData::FullBlock(block) => {
					self.chain_listener.block_connected(block, height);
				},
				BlockData::HeaderOnly(header) => {
					self.chain_listener.filtered_block_connected(header, &[], height);
				},
			}

			self.header_cache.block_connected(header.block_hash, header);
			new_tip = header;
		}

		Ok(())
	}
}

#[cfg(test)]
mod spv_client_tests {
	use super::*;
	use crate::test_utils::{Blockchain, NullChainListener};

	use bitcoin::network::Network;

	#[tokio::test]
	async fn poll_from_chain_without_headers() {
		let mut chain = Blockchain::default().with_height(3).without_headers();
		let best_tip = chain.at_height(1);

		let poller = poll::ChainPoller::new(&mut chain, Network::Testnet);
		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 {
			Err(e) => {
				assert_eq!(e.kind(), BlockSourceErrorKind::Persistent);
				assert_eq!(e.into_inner().as_ref().to_string(), "header not found");
			},
			Ok(_) => panic!("Expected error"),
		}
		assert_eq!(client.chain_tip, best_tip);
	}

	#[tokio::test]
	async fn poll_from_chain_with_common_tip() {
		let mut chain = Blockchain::default().with_height(3);
		let common_tip = chain.tip();

		let poller = poll::ChainPoller::new(&mut chain, Network::Testnet);
		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 {
			Err(e) => panic!("Unexpected error: {:?}", e),
			Ok((chain_tip, blocks_connected)) => {
				assert_eq!(chain_tip, ChainTip::Common);
				assert!(!blocks_connected);
			},
		}
		assert_eq!(client.chain_tip, common_tip);
	}

	#[tokio::test]
	async fn poll_from_chain_with_better_tip() {
		let mut chain = Blockchain::default().with_height(3);
		let new_tip = chain.tip();
		let old_tip = chain.at_height(1);

		let poller = poll::ChainPoller::new(&mut chain, Network::Testnet);
		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 {
			Err(e) => panic!("Unexpected error: {:?}", e),
			Ok((chain_tip, blocks_connected)) => {
				assert_eq!(chain_tip, ChainTip::Better(new_tip));
				assert!(blocks_connected);
			},
		}
		assert_eq!(client.chain_tip, new_tip);
	}

	#[tokio::test]
	async fn poll_from_chain_with_better_tip_and_without_any_new_blocks() {
		let mut chain = Blockchain::default().with_height(3).without_blocks(2..);
		let new_tip = chain.tip();
		let old_tip = chain.at_height(1);

		let poller = poll::ChainPoller::new(&mut chain, Network::Testnet);
		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 {
			Err(e) => panic!("Unexpected error: {:?}", e),
			Ok((chain_tip, blocks_connected)) => {
				assert_eq!(chain_tip, ChainTip::Better(new_tip));
				assert!(!blocks_connected);
			},
		}
		assert_eq!(client.chain_tip, old_tip);
	}

	#[tokio::test]
	async fn poll_from_chain_with_better_tip_and_without_some_new_blocks() {
		let mut chain = Blockchain::default().with_height(3).without_blocks(3..);
		let new_tip = chain.tip();
		let old_tip = chain.at_height(1);

		let poller = poll::ChainPoller::new(&mut chain, Network::Testnet);
		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 {
			Err(e) => panic!("Unexpected error: {:?}", e),
			Ok((chain_tip, blocks_connected)) => {
				assert_eq!(chain_tip, ChainTip::Better(new_tip));
				assert!(blocks_connected);
			},
		}
		assert_eq!(client.chain_tip, chain.at_height(2));
	}

	#[tokio::test]
	async fn poll_from_chain_with_worse_tip() {
		let mut chain = Blockchain::default().with_height(3);
		let best_tip = chain.tip();
		chain.disconnect_tip();
		let worse_tip = chain.tip();

		let poller = poll::ChainPoller::new(&mut chain, Network::Testnet);
		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 {
			Err(e) => panic!("Unexpected error: {:?}", e),
			Ok((chain_tip, blocks_connected)) => {
				assert_eq!(chain_tip, ChainTip::Worse(worse_tip));
				assert!(!blocks_connected);
			},
		}
		assert_eq!(client.chain_tip, best_tip);
	}
}

#[cfg(test)]
mod chain_notifier_tests {
	use super::*;
	use crate::test_utils::{Blockchain, MockChainListener};

	use bitcoin::network::Network;

	#[tokio::test]
	async fn sync_from_same_chain() {
		let mut chain = Blockchain::default().with_height(3);

		let new_tip = chain.tip();
		let old_tip = chain.at_height(1);
		let chain_listener = &MockChainListener::new()
			.expect_block_connected(*chain.at_height(2))
			.expect_block_connected(*new_tip);
		let mut notifier =
			ChainNotifier { header_cache: &mut chain.header_cache(0..=1), chain_listener };
		let mut poller = poll::ChainPoller::new(&mut chain, Network::Testnet);
		match notifier.synchronize_listener(new_tip, &old_tip, &mut poller).await {
			Err((e, _)) => panic!("Unexpected error: {:?}", e),
			Ok(_) => {},
		}
	}

	#[tokio::test]
	async fn sync_from_different_chains() {
		let mut test_chain = Blockchain::with_network(Network::Testnet).with_height(1);
		let main_chain = Blockchain::with_network(Network::Bitcoin).with_height(1);

		let new_tip = test_chain.tip();
		let old_tip = main_chain.tip();
		let chain_listener = &MockChainListener::new();
		let mut notifier =
			ChainNotifier { header_cache: &mut main_chain.header_cache(0..=1), chain_listener };
		let mut poller = poll::ChainPoller::new(&mut test_chain, Network::Testnet);
		match notifier.synchronize_listener(new_tip, &old_tip, &mut poller).await {
			Err((e, _)) => {
				assert_eq!(e.kind(), BlockSourceErrorKind::Persistent);
				assert_eq!(e.into_inner().as_ref().to_string(), "genesis block reached");
			},
			Ok(_) => panic!("Expected error"),
		}
	}

	#[tokio::test]
	async fn sync_from_equal_length_fork() {
		let main_chain = Blockchain::default().with_height(2);
		let mut fork_chain = main_chain.fork_at_height(1);

		let new_tip = fork_chain.tip();
		let old_tip = main_chain.tip();
		let chain_listener = &MockChainListener::new()
			.expect_blocks_disconnected(*fork_chain.at_height(1))
			.expect_block_connected(*new_tip);
		let mut notifier =
			ChainNotifier { header_cache: &mut main_chain.header_cache(0..=2), chain_listener };
		let mut poller = poll::ChainPoller::new(&mut fork_chain, Network::Testnet);
		match notifier.synchronize_listener(new_tip, &old_tip, &mut poller).await {
			Err((e, _)) => panic!("Unexpected error: {:?}", e),
			Ok(_) => {},
		}
	}

	#[tokio::test]
	async fn sync_from_shorter_fork() {
		let main_chain = Blockchain::default().with_height(3);
		let mut fork_chain = main_chain.fork_at_height(1);
		fork_chain.disconnect_tip();

		let new_tip = fork_chain.tip();
		let old_tip = main_chain.tip();
		let chain_listener = &MockChainListener::new()
			.expect_blocks_disconnected(*main_chain.at_height(1))
			.expect_block_connected(*new_tip);
		let mut notifier =
			ChainNotifier { header_cache: &mut main_chain.header_cache(0..=3), chain_listener };
		let mut poller = poll::ChainPoller::new(&mut fork_chain, Network::Testnet);
		match notifier.synchronize_listener(new_tip, &old_tip, &mut poller).await {
			Err((e, _)) => panic!("Unexpected error: {:?}", e),
			Ok(_) => {},
		}
	}

	#[tokio::test]
	async fn sync_from_longer_fork() {
		let mut main_chain = Blockchain::default().with_height(3);
		let mut fork_chain = main_chain.fork_at_height(1);
		main_chain.disconnect_tip();

		let new_tip = fork_chain.tip();
		let old_tip = main_chain.tip();
		let chain_listener = &MockChainListener::new()
			.expect_blocks_disconnected(*fork_chain.at_height(1))
			.expect_block_connected(*fork_chain.at_height(2))
			.expect_block_connected(*new_tip);
		let mut notifier =
			ChainNotifier { header_cache: &mut main_chain.header_cache(0..=2), chain_listener };
		let mut poller = poll::ChainPoller::new(&mut fork_chain, Network::Testnet);
		match notifier.synchronize_listener(new_tip, &old_tip, &mut poller).await {
			Err((e, _)) => panic!("Unexpected error: {:?}", e),
			Ok(_) => {},
		}
	}

	#[tokio::test]
	async fn sync_from_chain_without_headers() {
		let mut chain = Blockchain::default().with_height(3).without_headers();

		let new_tip = chain.tip();
		let old_tip = chain.at_height(1);
		let chain_listener = &MockChainListener::new();
		let mut notifier =
			ChainNotifier { header_cache: &mut chain.header_cache(0..=1), chain_listener };
		let mut poller = poll::ChainPoller::new(&mut chain, Network::Testnet);
		match notifier.synchronize_listener(new_tip, &old_tip, &mut poller).await {
			Err((_, tip)) => assert_eq!(tip, None),
			Ok(_) => panic!("Expected error"),
		}
	}

	#[tokio::test]
	async fn sync_from_chain_without_any_new_blocks() {
		let mut chain = Blockchain::default().with_height(3).without_blocks(2..);

		let new_tip = chain.tip();
		let old_tip = chain.at_height(1);
		let chain_listener = &MockChainListener::new();
		let mut notifier =
			ChainNotifier { header_cache: &mut chain.header_cache(0..=3), chain_listener };
		let mut poller = poll::ChainPoller::new(&mut chain, Network::Testnet);
		match notifier.synchronize_listener(new_tip, &old_tip, &mut poller).await {
			Err((_, tip)) => assert_eq!(tip, Some(old_tip)),
			Ok(_) => panic!("Expected error"),
		}
	}

	#[tokio::test]
	async fn sync_from_chain_without_some_new_blocks() {
		let mut chain = Blockchain::default().with_height(3).without_blocks(3..);

		let new_tip = chain.tip();
		let old_tip = chain.at_height(1);
		let chain_listener = &MockChainListener::new().expect_block_connected(*chain.at_height(2));
		let mut notifier =
			ChainNotifier { header_cache: &mut chain.header_cache(0..=3), chain_listener };
		let mut poller = poll::ChainPoller::new(&mut chain, Network::Testnet);
		match notifier.synchronize_listener(new_tip, &old_tip, &mut poller).await {
			Err((_, tip)) => assert_eq!(tip, Some(chain.at_height(2))),
			Ok(_) => panic!("Expected error"),
		}
	}

	#[tokio::test]
	async fn sync_from_chain_with_filtered_blocks() {
		let mut chain = Blockchain::default().with_height(3).filtered_blocks();

		let new_tip = chain.tip();
		let old_tip = chain.at_height(1);
		let chain_listener = &MockChainListener::new()
			.expect_filtered_block_connected(*chain.at_height(2))
			.expect_filtered_block_connected(*new_tip);
		let mut notifier =
			ChainNotifier { header_cache: &mut chain.header_cache(0..=1), chain_listener };
		let mut poller = poll::ChainPoller::new(&mut chain, Network::Testnet);
		match notifier.synchronize_listener(new_tip, &old_tip, &mut poller).await {
			Err((e, _)) => panic!("Unexpected error: {:?}", e),
			Ok(_) => {},
		}
	}
}