--- /dev/null
+use crate::{BlockSource, BlockSourceResult, Cache, ChainNotifier};
+use crate::poll::{ChainPoller, Validate, ValidatedBlockHeader};
+
+use bitcoin::blockdata::block::{Block, BlockHeader};
+use bitcoin::hash_types::BlockHash;
+use bitcoin::network::constants::Network;
+
+use lightning::chain;
+
+/// 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.
+///
+/// Useful during startup to bring the [`ChannelManager`] and each [`ChannelMonitor`] in sync before
+/// switching to [`SpvClient`]. For example:
+///
+/// ```
+/// use bitcoin::hash_types::BlockHash;
+/// use bitcoin::network::constants::Network;
+///
+/// use lightning::chain;
+/// use lightning::chain::Watch;
+/// use lightning::chain::chainmonitor::ChainMonitor;
+/// use lightning::chain::channelmonitor;
+/// use lightning::chain::channelmonitor::ChannelMonitor;
+/// use lightning::chain::chaininterface::BroadcasterInterface;
+/// use lightning::chain::chaininterface::FeeEstimator;
+/// use lightning::chain::keysinterface;
+/// use lightning::chain::keysinterface::KeysInterface;
+/// use lightning::ln::channelmanager::ChannelManager;
+/// use lightning::ln::channelmanager::ChannelManagerReadArgs;
+/// use lightning::util::config::UserConfig;
+/// use lightning::util::logger::Logger;
+/// use lightning::util::ser::ReadableArgs;
+///
+/// use lightning_block_sync::*;
+///
+/// use std::cell::RefCell;
+/// use std::io::Cursor;
+///
+/// async fn init_sync<
+/// B: BlockSource,
+/// K: KeysInterface<Signer = S>,
+/// S: keysinterface::Sign,
+/// T: BroadcasterInterface,
+/// F: FeeEstimator,
+/// L: Logger,
+/// C: chain::Filter,
+/// P: channelmonitor::Persist<S>,
+/// >(
+/// block_source: &mut B,
+/// chain_monitor: &ChainMonitor<S, &C, &T, &F, &L, &P>,
+/// config: UserConfig,
+/// keys_manager: &K,
+/// tx_broadcaster: &T,
+/// fee_estimator: &F,
+/// logger: &L,
+/// persister: &P,
+/// ) {
+/// // Read a serialized channel monitor paired with the block hash when it was persisted.
+/// let serialized_monitor = "...";
+/// let (monitor_block_hash, mut monitor) = <(BlockHash, ChannelMonitor<S>)>::read(
+/// &mut Cursor::new(&serialized_monitor), keys_manager).unwrap();
+///
+/// // Read the channel manager paired with the block hash when it was persisted.
+/// let serialized_manager = "...";
+/// let (manager_block_hash, mut manager) = {
+/// let read_args = ChannelManagerReadArgs::new(
+/// keys_manager,
+/// fee_estimator,
+/// chain_monitor,
+/// tx_broadcaster,
+/// logger,
+/// config,
+/// vec![&mut monitor],
+/// );
+/// <(BlockHash, ChannelManager<S, &ChainMonitor<S, &C, &T, &F, &L, &P>, &T, &K, &F, &L>)>::read(
+/// &mut Cursor::new(&serialized_manager), read_args).unwrap()
+/// };
+///
+/// // Synchronize any channel monitors and the channel manager to be on the best block.
+/// let mut cache = UnboundedCache::new();
+/// let mut monitor_listener = (RefCell::new(monitor), &*tx_broadcaster, &*fee_estimator, &*logger);
+/// let listeners = vec![
+/// (monitor_block_hash, &mut monitor_listener as &mut dyn chain::Listen),
+/// (manager_block_hash, &mut manager as &mut dyn chain::Listen),
+/// ];
+/// let chain_tip = init::synchronize_listeners(
+/// block_source, Network::Bitcoin, &mut cache, listeners).await.unwrap();
+///
+/// // Allow the chain monitor to watch any channels.
+/// let monitor = monitor_listener.0.into_inner();
+/// chain_monitor.watch_channel(monitor.get_funding_txo().0, monitor);
+///
+/// // 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);
+/// }
+/// ```
+///
+/// [`SpvClient`]: ../struct.SpvClient.html
+/// [`ChannelManager`]: ../../lightning/ln/channelmanager/struct.ChannelManager.html
+/// [`ChannelMonitor`]: ../../lightning/chain/channelmonitor/struct.ChannelMonitor.html
+pub async fn synchronize_listeners<B: BlockSource, C: Cache>(
+ block_source: &mut B,
+ network: Network,
+ header_cache: &mut C,
+ mut chain_listeners: Vec<(BlockHash, &mut dyn chain::Listen)>,
+) -> BlockSourceResult<ValidatedBlockHeader> {
+ let (best_block_hash, best_block_height) = block_source.get_best_block().await?;
+ let best_header = block_source
+ .get_header(&best_block_hash, best_block_height).await?
+ .validate(best_block_hash)?;
+
+ // Fetch the header for the block hash paired with each listener.
+ let mut chain_listeners_with_old_headers = Vec::new();
+ for (old_block_hash, chain_listener) in chain_listeners.drain(..) {
+ let old_header = match header_cache.look_up(&old_block_hash) {
+ Some(header) => *header,
+ None => block_source
+ .get_header(&old_block_hash, None).await?
+ .validate(old_block_hash)?
+ };
+ chain_listeners_with_old_headers.push((old_header, chain_listener))
+ }
+
+ // Find differences and disconnect blocks for each listener individually.
+ let mut chain_poller = ChainPoller::new(block_source, network);
+ let mut chain_listeners_at_height = Vec::new();
+ let mut most_common_ancestor = None;
+ let mut most_connected_blocks = Vec::new();
+ for (old_header, chain_listener) in chain_listeners_with_old_headers.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 difference =
+ chain_notifier.find_difference(best_header, &old_header, &mut chain_poller).await?;
+ chain_notifier.disconnect_blocks(difference.disconnected_blocks);
+ (difference.common_ancestor, difference.connected_blocks)
+ };
+
+ // Keep track of the most common ancestor and all blocks connected across all listeners.
+ chain_listeners_at_height.push((common_ancestor.height, chain_listener));
+ if connected_blocks.len() > most_connected_blocks.len() {
+ most_common_ancestor = Some(common_ancestor);
+ most_connected_blocks = connected_blocks;
+ }
+ }
+
+ // 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 };
+ chain_notifier.connect_blocks(common_ancestor, most_connected_blocks, &mut chain_poller)
+ .await.or_else(|(e, _)| Err(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 block_disconnected(&mut self, _block_hash: &BlockHash) -> Option<ValidatedBlockHeader> {
+ None
+ }
+}
+
+/// Wrapper for supporting dynamically sized chain listeners.
+struct DynamicChainListener<'a>(&'a mut dyn chain::Listen);
+
+impl<'a> chain::Listen for DynamicChainListener<'a> {
+ fn block_connected(&self, _block: &Block, _height: u32) {
+ unreachable!()
+ }
+
+ fn block_disconnected(&self, header: &BlockHeader, height: u32) {
+ self.0.block_disconnected(header, height)
+ }
+}
+
+/// A set of dynamically sized chain listeners, each paired with a starting block height.
+struct ChainListenerSet<'a>(Vec<(u32, &'a mut dyn chain::Listen)>);
+
+impl<'a> chain::Listen for ChainListenerSet<'a> {
+ fn block_connected(&self, block: &Block, height: u32) {
+ for (starting_height, chain_listener) in self.0.iter() {
+ if height > *starting_height {
+ chain_listener.block_connected(block, height);
+ }
+ }
+ }
+
+ fn block_disconnected(&self, _header: &BlockHeader, _height: u32) {
+ unreachable!()
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use crate::test_utils::{Blockchain, MockChainListener};
+ use super::*;
+
+ use bitcoin::network::constants::Network;
+
+ #[tokio::test]
+ async fn sync_from_same_chain() {
+ let mut chain = Blockchain::default().with_height(4);
+
+ let mut listener_1 = MockChainListener::new()
+ .expect_block_connected(*chain.at_height(2))
+ .expect_block_connected(*chain.at_height(3))
+ .expect_block_connected(*chain.at_height(4));
+ let mut listener_2 = MockChainListener::new()
+ .expect_block_connected(*chain.at_height(3))
+ .expect_block_connected(*chain.at_height(4));
+ let mut listener_3 = MockChainListener::new()
+ .expect_block_connected(*chain.at_height(4));
+
+ let listeners = vec![
+ (chain.at_height(1).block_hash, &mut listener_1 as &mut dyn chain::Listen),
+ (chain.at_height(2).block_hash, &mut listener_2 as &mut dyn chain::Listen),
+ (chain.at_height(3).block_hash, &mut listener_3 as &mut dyn chain::Listen),
+ ];
+ let mut cache = chain.header_cache(0..=4);
+ match synchronize_listeners(&mut chain, Network::Bitcoin, &mut cache, listeners).await {
+ Ok(header) => assert_eq!(header, chain.tip()),
+ Err(e) => panic!("Unexpected error: {:?}", e),
+ }
+ }
+
+ #[tokio::test]
+ async fn sync_from_different_chains() {
+ let mut main_chain = Blockchain::default().with_height(4);
+ let fork_chain_1 = main_chain.fork_at_height(1);
+ let fork_chain_2 = main_chain.fork_at_height(2);
+ let fork_chain_3 = main_chain.fork_at_height(3);
+
+ let mut listener_1 = MockChainListener::new()
+ .expect_block_disconnected(*fork_chain_1.at_height(4))
+ .expect_block_disconnected(*fork_chain_1.at_height(3))
+ .expect_block_disconnected(*fork_chain_1.at_height(2))
+ .expect_block_connected(*main_chain.at_height(2))
+ .expect_block_connected(*main_chain.at_height(3))
+ .expect_block_connected(*main_chain.at_height(4));
+ let mut listener_2 = MockChainListener::new()
+ .expect_block_disconnected(*fork_chain_2.at_height(4))
+ .expect_block_disconnected(*fork_chain_2.at_height(3))
+ .expect_block_connected(*main_chain.at_height(3))
+ .expect_block_connected(*main_chain.at_height(4));
+ let mut listener_3 = MockChainListener::new()
+ .expect_block_disconnected(*fork_chain_3.at_height(4))
+ .expect_block_connected(*main_chain.at_height(4));
+
+ let listeners = vec![
+ (fork_chain_1.tip().block_hash, &mut listener_1 as &mut dyn chain::Listen),
+ (fork_chain_2.tip().block_hash, &mut listener_2 as &mut dyn chain::Listen),
+ (fork_chain_3.tip().block_hash, &mut listener_3 as &mut 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(&mut 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 sync_from_overlapping_chains() {
+ let mut main_chain = Blockchain::default().with_height(4);
+ let fork_chain_1 = main_chain.fork_at_height(1);
+ let fork_chain_2 = fork_chain_1.fork_at_height(2);
+ let fork_chain_3 = fork_chain_2.fork_at_height(3);
+
+ let mut listener_1 = MockChainListener::new()
+ .expect_block_disconnected(*fork_chain_1.at_height(4))
+ .expect_block_disconnected(*fork_chain_1.at_height(3))
+ .expect_block_disconnected(*fork_chain_1.at_height(2))
+ .expect_block_connected(*main_chain.at_height(2))
+ .expect_block_connected(*main_chain.at_height(3))
+ .expect_block_connected(*main_chain.at_height(4));
+ let mut listener_2 = MockChainListener::new()
+ .expect_block_disconnected(*fork_chain_2.at_height(4))
+ .expect_block_disconnected(*fork_chain_2.at_height(3))
+ .expect_block_disconnected(*fork_chain_2.at_height(2))
+ .expect_block_connected(*main_chain.at_height(2))
+ .expect_block_connected(*main_chain.at_height(3))
+ .expect_block_connected(*main_chain.at_height(4));
+ let mut listener_3 = MockChainListener::new()
+ .expect_block_disconnected(*fork_chain_3.at_height(4))
+ .expect_block_disconnected(*fork_chain_3.at_height(3))
+ .expect_block_disconnected(*fork_chain_3.at_height(2))
+ .expect_block_connected(*main_chain.at_height(2))
+ .expect_block_connected(*main_chain.at_height(3))
+ .expect_block_connected(*main_chain.at_height(4));
+
+ let listeners = vec![
+ (fork_chain_1.tip().block_hash, &mut listener_1 as &mut dyn chain::Listen),
+ (fork_chain_2.tip().block_hash, &mut listener_2 as &mut dyn chain::Listen),
+ (fork_chain_3.tip().block_hash, &mut listener_3 as &mut 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(&mut 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 mut main_chain = Blockchain::default().with_height(2);
+ let fork_chain = main_chain.fork_at_height(1);
+ let new_tip = main_chain.tip();
+ let old_tip = fork_chain.tip();
+
+ let mut listener = MockChainListener::new()
+ .expect_block_disconnected(*old_tip)
+ .expect_block_connected(*new_tip);
+
+ let listeners = vec![(old_tip.block_hash, &mut listener as &mut dyn chain::Listen)];
+ let mut cache = fork_chain.header_cache(2..=2);
+ match synchronize_listeners(&mut main_chain, Network::Bitcoin, &mut cache, listeners).await {
+ Ok(_) => {
+ assert!(cache.contains_key(&new_tip.block_hash));
+ assert!(cache.contains_key(&old_tip.block_hash));
+ },
+ Err(e) => panic!("Unexpected error: {:?}", e),
+ }
+ }
+}
//! 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.
//!
//! 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.
//!
+//! [`SpvClient`]: struct.SpvClient.html
//! [`BlockSource`]: trait.BlockSource.html
#[cfg(any(feature = "rest-client", feature = "rpc-client"))]
pub mod http;
+pub mod init;
+pub mod poll;
+
#[cfg(feature = "rest-client")]
pub mod rest;
#[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;
+use crate::poll::{ChainTip, Poll, ValidatedBlockHeader};
+
use bitcoin::blockdata::block::{Block, BlockHeader};
use bitcoin::hash_types::BlockHash;
use bitcoin::util::uint::Uint256;
+use lightning::chain;
+use lightning::chain::Listen;
+
use std::future::Future;
+use std::ops::Deref;
use std::pin::Pin;
/// Abstract type for retrieving block headers and data.
/// error.
fn get_block<'a>(&'a mut self, header_hash: &'a BlockHash) -> AsyncBlockSourceResult<'a, Block>;
- // TODO: Phrase in terms of `Poll` once added.
- /// Returns the hash of the best block and, optionally, its height. When polling a block source,
- /// the height is passed to `get_header` to allow for a more efficient lookup.
+ /// 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.
+ ///
+ /// [`Poll`]: poll/trait.Poll.html
+ /// [`get_header`]: #tymethod.get_header
fn get_best_block<'a>(&'a mut self) -> AsyncBlockSourceResult<(BlockHash, Option<u32>)>;
}
///
/// 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)]
+#[derive(Clone, Copy, Debug, PartialEq)]
pub enum BlockSourceErrorKind {
/// Indicates an error that won't resolve when retrying a request (e.g., invalid data).
Persistent,
/// of equivalent weight.
pub chainwork: Uint256,
}
+
+/// 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.
+///
+/// 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>
+where L::Target: chain::Listen {
+ chain_tip: ValidatedBlockHeader,
+ chain_poller: P,
+ chain_notifier: ChainNotifier<'a, C, L>,
+}
+
+/// The `Cache` trait defines behavior for managing a block header cache, where block headers are
+/// keyed by block hash.
+///
+/// Used by [`ChainNotifier`] to store headers along the best chain, which is important for ensuring
+/// that blocks can be disconnected if they are no longer accessible from a block source (e.g., if
+/// the block source does not store stale forks indefinitely).
+///
+/// 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.
+///
+/// [`ChainNotifier`]: struct.ChainNotifier.html
+pub trait Cache {
+ /// Retrieves the block header keyed by the given block hash.
+ fn look_up(&self, block_hash: &BlockHash) -> Option<&ValidatedBlockHeader>;
+
+ /// Called when a block has been connected to the best chain to ensure it is available to be
+ /// disconnected later if needed.
+ fn block_connected(&mut self, block_hash: BlockHash, block_header: ValidatedBlockHeader);
+
+ /// Called when a block has been disconnected from the best chain. Once disconnected, a block's
+ /// header is no longer needed and thus can be removed.
+ fn block_disconnected(&mut self, block_hash: &BlockHash) -> Option<ValidatedBlockHeader>;
+}
+
+/// Unbounded cache of block headers keyed by block hash.
+pub type UnboundedCache = std::collections::HashMap<BlockHash, ValidatedBlockHeader>;
+
+impl Cache for 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 block_disconnected(&mut self, block_hash: &BlockHash) -> Option<ValidatedBlockHeader> {
+ self.remove(block_hash)
+ }
+}
+
+impl<'a, P: Poll, C: Cache, L: Deref> SpvClient<'a, P, C, 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`]: struct.SpvClient.html#method.poll_best_tip
+ pub fn new(
+ chain_tip: ValidatedBlockHeader,
+ chain_poller: P,
+ header_cache: &'a mut C,
+ chain_listener: L,
+ ) -> Self {
+ let chain_notifier = ChainNotifier { header_cache, chain_listener };
+ Self { chain_tip, chain_poller, chain_notifier }
+ }
+
+ /// 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 {
+ match self.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/trait.Listen.html
+pub struct ChainNotifier<'a, 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,
+
+ /// Listener that will be notified of connected or disconnected blocks.
+ chain_listener: 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 disconnected from the chain since the last poll.
+ disconnected_blocks: Vec<ValidatedBlockHeader>,
+
+ /// Blocks that were connected to the chain since the last poll.
+ connected_blocks: Vec<ValidatedBlockHeader>,
+}
+
+impl<'a, C: Cache, L: Deref> ChainNotifier<'a, C, L> where L::Target: chain::Listen {
+ /// 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(new_header, old_header, chain_poller).await
+ .map_err(|e| (e, None))?;
+ self.disconnect_blocks(difference.disconnected_blocks);
+ 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_header` as its tip.
+ ///
+ /// Walks backwards from `current_header` and `prev_header`, finding the common ancestor.
+ async fn find_difference<P: Poll>(
+ &self,
+ current_header: ValidatedBlockHeader,
+ prev_header: &ValidatedBlockHeader,
+ chain_poller: &mut P,
+ ) -> BlockSourceResult<ChainDifference> {
+ let mut disconnected_blocks = Vec::new();
+ 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 {
+ disconnected_blocks.push(previous);
+ 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, ¤t).await?;
+ }
+ }
+
+ let common_ancestor = current;
+ Ok(ChainDifference { common_ancestor, disconnected_blocks, 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, mut disconnected_blocks: Vec<ValidatedBlockHeader>) {
+ for header in disconnected_blocks.drain(..) {
+ if let Some(cached_header) = self.header_cache.block_disconnected(&header.block_hash) {
+ assert_eq!(cached_header, header);
+ }
+ self.chain_listener.block_disconnected(&header.header, header.height);
+ }
+ }
+
+ /// 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 block = chain_poller
+ .fetch_block(&header).await
+ .or_else(|e| Err((e, Some(new_tip))))?;
+ debug_assert_eq!(block.block_hash, header.block_hash);
+
+ self.header_cache.block_connected(header.block_hash, header);
+ self.chain_listener.block_connected(&block, header.height);
+ new_tip = header;
+ }
+
+ Ok(())
+ }
+}
+
+#[cfg(test)]
+mod spv_client_tests {
+ use crate::test_utils::{Blockchain, NullChainListener};
+ use super::*;
+
+ use bitcoin::network::constants::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 mut cache = UnboundedCache::new();
+ let mut listener = NullChainListener {};
+ let mut client = SpvClient::new(best_tip, poller, &mut 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 mut cache = UnboundedCache::new();
+ let mut listener = NullChainListener {};
+ let mut client = SpvClient::new(common_tip, poller, &mut 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 mut cache = UnboundedCache::new();
+ let mut listener = NullChainListener {};
+ let mut client = SpvClient::new(old_tip, poller, &mut 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 mut cache = UnboundedCache::new();
+ let mut listener = NullChainListener {};
+ let mut client = SpvClient::new(old_tip, poller, &mut 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 mut cache = UnboundedCache::new();
+ let mut listener = NullChainListener {};
+ let mut client = SpvClient::new(old_tip, poller, &mut 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 mut cache = UnboundedCache::new();
+ let mut listener = NullChainListener {};
+ let mut client = SpvClient::new(best_tip, poller, &mut 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 crate::test_utils::{Blockchain, MockChainListener};
+ use super::*;
+
+ use bitcoin::network::constants::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_block_disconnected(*old_tip)
+ .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_block_disconnected(*old_tip)
+ .expect_block_disconnected(*main_chain.at_height(2))
+ .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_block_disconnected(*old_tip)
+ .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"),
+ }
+ }
+}
--- /dev/null
+use crate::{AsyncBlockSourceResult, BlockHeaderData, BlockSource, BlockSourceError, BlockSourceResult};
+
+use bitcoin::blockdata::block::Block;
+use bitcoin::hash_types::BlockHash;
+use bitcoin::network::constants::Network;
+
+use std::ops::DerefMut;
+
+/// The `Poll` trait defines behavior for polling block sources for a chain tip and retrieving
+/// related chain data. It serves as an adapter for `BlockSource`.
+///
+/// [`ChainPoller`] adapts a single `BlockSource`, while any other implementations of `Poll` are
+/// required to be built in terms of it to ensure chain data validity.
+///
+/// [`ChainPoller`]: ../struct.ChainPoller.html
+pub trait Poll {
+ /// Returns a chain tip in terms of its relationship to the provided chain tip.
+ fn poll_chain_tip<'a>(&'a mut self, best_known_chain_tip: ValidatedBlockHeader) ->
+ AsyncBlockSourceResult<'a, ChainTip>;
+
+ /// Returns the header that preceded the given header in the chain.
+ fn look_up_previous_header<'a>(&'a mut self, header: &'a ValidatedBlockHeader) ->
+ AsyncBlockSourceResult<'a, ValidatedBlockHeader>;
+
+ /// Returns the block associated with the given header.
+ fn fetch_block<'a>(&'a mut self, header: &'a ValidatedBlockHeader) ->
+ AsyncBlockSourceResult<'a, ValidatedBlock>;
+}
+
+/// A chain tip relative to another chain tip in terms of block hash and chainwork.
+#[derive(Clone, Debug, PartialEq)]
+pub enum ChainTip {
+ /// A chain tip with the same hash as another chain's tip.
+ Common,
+
+ /// A chain tip with more chainwork than another chain's tip.
+ Better(ValidatedBlockHeader),
+
+ /// A chain tip with less or equal chainwork than another chain's tip. In either case, the
+ /// hashes of each tip will be different.
+ Worse(ValidatedBlockHeader),
+}
+
+/// The `Validate` trait defines behavior for validating chain data.
+pub(crate) trait Validate {
+ /// The validated data wrapper which can be dereferenced to obtain the validated data.
+ type T: std::ops::Deref<Target = Self>;
+
+ /// Validates the chain data against the given block hash and any criteria needed to ensure that
+ /// it is internally consistent.
+ fn validate(self, block_hash: BlockHash) -> BlockSourceResult<Self::T>;
+}
+
+impl Validate for BlockHeaderData {
+ type T = ValidatedBlockHeader;
+
+ fn validate(self, block_hash: BlockHash) -> BlockSourceResult<Self::T> {
+ self.header
+ .validate_pow(&self.header.target())
+ .or_else(|e| Err(BlockSourceError::persistent(e)))?;
+
+ // TODO: Use the result of validate_pow instead of recomputing the block hash once upstream.
+ if self.header.block_hash() != block_hash {
+ return Err(BlockSourceError::persistent("invalid block hash"));
+ }
+
+ Ok(ValidatedBlockHeader { block_hash, inner: self })
+ }
+}
+
+impl Validate for Block {
+ type T = ValidatedBlock;
+
+ fn validate(self, block_hash: BlockHash) -> BlockSourceResult<Self::T> {
+ self.header
+ .validate_pow(&self.header.target())
+ .or_else(|e| Err(BlockSourceError::persistent(e)))?;
+
+ // TODO: Use the result of validate_pow instead of recomputing the block hash once upstream.
+ if self.block_hash() != block_hash {
+ return Err(BlockSourceError::persistent("invalid block hash"));
+ }
+
+ if !self.check_merkle_root() {
+ return Err(BlockSourceError::persistent("invalid merkle root"));
+ }
+
+ if !self.check_witness_commitment() {
+ return Err(BlockSourceError::persistent("invalid witness commitment"));
+ }
+
+ Ok(ValidatedBlock { block_hash, inner: self })
+ }
+}
+
+/// A block header with validated proof of work and corresponding block hash.
+#[derive(Clone, Copy, Debug, PartialEq)]
+pub struct ValidatedBlockHeader {
+ pub(crate) block_hash: BlockHash,
+ inner: BlockHeaderData,
+}
+
+impl std::ops::Deref for ValidatedBlockHeader {
+ type Target = BlockHeaderData;
+
+ fn deref(&self) -> &Self::Target {
+ &self.inner
+ }
+}
+
+impl ValidatedBlockHeader {
+ /// Checks that the header correctly builds on previous_header: the claimed work differential
+ /// matches the actual PoW and the difficulty transition is possible, i.e., within 4x.
+ fn check_builds_on(&self, previous_header: &ValidatedBlockHeader, network: Network) -> BlockSourceResult<()> {
+ if self.header.prev_blockhash != previous_header.block_hash {
+ return Err(BlockSourceError::persistent("invalid previous block hash"));
+ }
+
+ if self.height != previous_header.height + 1 {
+ return Err(BlockSourceError::persistent("invalid block height"));
+ }
+
+ let work = self.header.work();
+ if self.chainwork != previous_header.chainwork + work {
+ return Err(BlockSourceError::persistent("invalid chainwork"));
+ }
+
+ if let Network::Bitcoin = network {
+ if self.height % 2016 == 0 {
+ let previous_work = previous_header.header.work();
+ if work > (previous_work << 2) || work < (previous_work >> 2) {
+ return Err(BlockSourceError::persistent("invalid difficulty transition"))
+ }
+ } else if self.header.bits != previous_header.header.bits {
+ return Err(BlockSourceError::persistent("invalid difficulty"))
+ }
+ }
+
+ Ok(())
+ }
+}
+
+/// A block with validated data against its transaction list and corresponding block hash.
+pub struct ValidatedBlock {
+ pub(crate) block_hash: BlockHash,
+ inner: Block,
+}
+
+impl std::ops::Deref for ValidatedBlock {
+ type Target = Block;
+
+ fn deref(&self) -> &Self::Target {
+ &self.inner
+ }
+}
+
+/// The canonical `Poll` implementation used for a single `BlockSource`.
+///
+/// Other `Poll` implementations must be built using `ChainPoller` as it provides the only means of
+/// validating chain data.
+pub struct ChainPoller<B: DerefMut<Target=T> + Sized + Sync + Send, T: BlockSource> {
+ block_source: B,
+ network: Network,
+}
+
+impl<B: DerefMut<Target=T> + Sized + Sync + Send, T: BlockSource> ChainPoller<B, T> {
+ /// Creates a new poller for the given block source.
+ ///
+ /// If the `network` parameter is mainnet, then the difficulty between blocks is checked for
+ /// validity.
+ pub fn new(block_source: B, network: Network) -> Self {
+ Self { block_source, network }
+ }
+}
+
+impl<B: DerefMut<Target=T> + Sized + Sync + Send, T: BlockSource> Poll for ChainPoller<B, T> {
+ fn poll_chain_tip<'a>(&'a mut self, best_known_chain_tip: ValidatedBlockHeader) ->
+ AsyncBlockSourceResult<'a, ChainTip>
+ {
+ Box::pin(async move {
+ let (block_hash, height) = self.block_source.get_best_block().await?;
+ if block_hash == best_known_chain_tip.header.block_hash() {
+ return Ok(ChainTip::Common);
+ }
+
+ let chain_tip = self.block_source
+ .get_header(&block_hash, height).await?
+ .validate(block_hash)?;
+ if chain_tip.chainwork > best_known_chain_tip.chainwork {
+ Ok(ChainTip::Better(chain_tip))
+ } else {
+ Ok(ChainTip::Worse(chain_tip))
+ }
+ })
+ }
+
+ fn look_up_previous_header<'a>(&'a mut self, header: &'a ValidatedBlockHeader) ->
+ AsyncBlockSourceResult<'a, ValidatedBlockHeader>
+ {
+ Box::pin(async move {
+ if header.height == 0 {
+ return Err(BlockSourceError::persistent("genesis block reached"));
+ }
+
+ let previous_hash = &header.header.prev_blockhash;
+ let height = header.height - 1;
+ let previous_header = self.block_source
+ .get_header(previous_hash, Some(height)).await?
+ .validate(*previous_hash)?;
+ header.check_builds_on(&previous_header, self.network)?;
+
+ Ok(previous_header)
+ })
+ }
+
+ fn fetch_block<'a>(&'a mut self, header: &'a ValidatedBlockHeader) ->
+ AsyncBlockSourceResult<'a, ValidatedBlock>
+ {
+ Box::pin(async move {
+ self.block_source
+ .get_block(&header.block_hash).await?
+ .validate(header.block_hash)
+ })
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use crate::*;
+ use crate::test_utils::Blockchain;
+ use super::*;
+ use bitcoin::util::uint::Uint256;
+
+ #[tokio::test]
+ async fn poll_empty_chain() {
+ let mut chain = Blockchain::default().with_height(0);
+ let best_known_chain_tip = chain.tip();
+ chain.disconnect_tip();
+
+ let mut poller = ChainPoller::new(&mut chain, Network::Bitcoin);
+ match poller.poll_chain_tip(best_known_chain_tip).await {
+ Err(e) => {
+ assert_eq!(e.kind(), BlockSourceErrorKind::Transient);
+ assert_eq!(e.into_inner().as_ref().to_string(), "empty chain");
+ },
+ Ok(_) => panic!("Expected error"),
+ }
+ }
+
+ #[tokio::test]
+ async fn poll_chain_without_headers() {
+ let mut chain = Blockchain::default().with_height(1).without_headers();
+ let best_known_chain_tip = chain.at_height(0);
+
+ let mut poller = ChainPoller::new(&mut chain, Network::Bitcoin);
+ match poller.poll_chain_tip(best_known_chain_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"),
+ }
+ }
+
+ #[tokio::test]
+ async fn poll_chain_with_invalid_pow() {
+ let mut chain = Blockchain::default().with_height(1);
+ let best_known_chain_tip = chain.at_height(0);
+
+ // Invalidate the tip by changing its target.
+ chain.blocks.last_mut().unwrap().header.bits =
+ BlockHeader::compact_target_from_u256(&Uint256::from_be_bytes([0; 32]));
+
+ let mut poller = ChainPoller::new(&mut chain, Network::Bitcoin);
+ match poller.poll_chain_tip(best_known_chain_tip).await {
+ Err(e) => {
+ assert_eq!(e.kind(), BlockSourceErrorKind::Persistent);
+ assert_eq!(e.into_inner().as_ref().to_string(), "block target correct but not attained");
+ },
+ Ok(_) => panic!("Expected error"),
+ }
+ }
+
+ #[tokio::test]
+ async fn poll_chain_with_malformed_headers() {
+ let mut chain = Blockchain::default().with_height(1).malformed_headers();
+ let best_known_chain_tip = chain.at_height(0);
+
+ let mut poller = ChainPoller::new(&mut chain, Network::Bitcoin);
+ match poller.poll_chain_tip(best_known_chain_tip).await {
+ Err(e) => {
+ assert_eq!(e.kind(), BlockSourceErrorKind::Persistent);
+ assert_eq!(e.into_inner().as_ref().to_string(), "invalid block hash");
+ },
+ Ok(_) => panic!("Expected error"),
+ }
+ }
+
+ #[tokio::test]
+ async fn poll_chain_with_common_tip() {
+ let mut chain = Blockchain::default().with_height(0);
+ let best_known_chain_tip = chain.tip();
+
+ let mut poller = ChainPoller::new(&mut chain, Network::Bitcoin);
+ match poller.poll_chain_tip(best_known_chain_tip).await {
+ Err(e) => panic!("Unexpected error: {:?}", e),
+ Ok(tip) => assert_eq!(tip, ChainTip::Common),
+ }
+ }
+
+ #[tokio::test]
+ async fn poll_chain_with_uncommon_tip_but_equal_chainwork() {
+ let mut chain = Blockchain::default().with_height(1);
+ let best_known_chain_tip = chain.tip();
+
+ // Change the nonce to get a different block hash with the same chainwork.
+ chain.blocks.last_mut().unwrap().header.nonce += 1;
+ let worse_chain_tip = chain.tip();
+ assert_eq!(best_known_chain_tip.chainwork, worse_chain_tip.chainwork);
+
+ let mut poller = ChainPoller::new(&mut chain, Network::Bitcoin);
+ match poller.poll_chain_tip(best_known_chain_tip).await {
+ Err(e) => panic!("Unexpected error: {:?}", e),
+ Ok(tip) => assert_eq!(tip, ChainTip::Worse(worse_chain_tip)),
+ }
+ }
+
+ #[tokio::test]
+ async fn poll_chain_with_worse_tip() {
+ let mut chain = Blockchain::default().with_height(1);
+ let best_known_chain_tip = chain.tip();
+
+ chain.disconnect_tip();
+ let worse_chain_tip = chain.tip();
+
+ let mut poller = ChainPoller::new(&mut chain, Network::Bitcoin);
+ match poller.poll_chain_tip(best_known_chain_tip).await {
+ Err(e) => panic!("Unexpected error: {:?}", e),
+ Ok(tip) => assert_eq!(tip, ChainTip::Worse(worse_chain_tip)),
+ }
+ }
+
+ #[tokio::test]
+ async fn poll_chain_with_better_tip() {
+ let mut chain = Blockchain::default().with_height(1);
+ let best_known_chain_tip = chain.at_height(0);
+
+ let better_chain_tip = chain.tip();
+
+ let mut poller = ChainPoller::new(&mut chain, Network::Bitcoin);
+ match poller.poll_chain_tip(best_known_chain_tip).await {
+ Err(e) => panic!("Unexpected error: {:?}", e),
+ Ok(tip) => assert_eq!(tip, ChainTip::Better(better_chain_tip)),
+ }
+ }
+}
--- /dev/null
+use crate::{AsyncBlockSourceResult, BlockHeaderData, BlockSource, BlockSourceError, UnboundedCache};
+use crate::poll::{Validate, ValidatedBlockHeader};
+
+use bitcoin::blockdata::block::{Block, BlockHeader};
+use bitcoin::blockdata::constants::genesis_block;
+use bitcoin::hash_types::BlockHash;
+use bitcoin::network::constants::Network;
+use bitcoin::util::uint::Uint256;
+
+use lightning::chain;
+
+use std::cell::RefCell;
+use std::collections::VecDeque;
+
+#[derive(Default)]
+pub struct Blockchain {
+ pub blocks: Vec<Block>,
+ without_blocks: Option<std::ops::RangeFrom<usize>>,
+ without_headers: bool,
+ malformed_headers: bool,
+}
+
+impl Blockchain {
+ pub fn default() -> Self {
+ Blockchain::with_network(Network::Bitcoin)
+ }
+
+ pub fn with_network(network: Network) -> Self {
+ let blocks = vec![genesis_block(network)];
+ Self { blocks, ..Default::default() }
+ }
+
+ pub fn with_height(mut self, height: usize) -> Self {
+ self.blocks.reserve_exact(height);
+ let bits = BlockHeader::compact_target_from_u256(&Uint256::from_be_bytes([0xff; 32]));
+ for i in 1..=height {
+ let prev_block = &self.blocks[i - 1];
+ let prev_blockhash = prev_block.block_hash();
+ let time = prev_block.header.time + height as u32;
+ self.blocks.push(Block {
+ header: BlockHeader {
+ version: 0,
+ prev_blockhash,
+ merkle_root: Default::default(),
+ time,
+ bits,
+ nonce: 0,
+ },
+ txdata: vec![],
+ });
+ }
+ self
+ }
+
+ pub fn without_blocks(self, range: std::ops::RangeFrom<usize>) -> Self {
+ Self { without_blocks: Some(range), ..self }
+ }
+
+ pub fn without_headers(self) -> Self {
+ Self { without_headers: true, ..self }
+ }
+
+ pub fn malformed_headers(self) -> Self {
+ Self { malformed_headers: true, ..self }
+ }
+
+ pub fn fork_at_height(&self, height: usize) -> Self {
+ assert!(height + 1 < self.blocks.len());
+ let mut blocks = self.blocks.clone();
+ let mut prev_blockhash = blocks[height].block_hash();
+ for block in blocks.iter_mut().skip(height + 1) {
+ block.header.prev_blockhash = prev_blockhash;
+ block.header.nonce += 1;
+ prev_blockhash = block.block_hash();
+ }
+ Self { blocks, without_blocks: None, ..*self }
+ }
+
+ pub fn at_height(&self, height: usize) -> ValidatedBlockHeader {
+ let block_header = self.at_height_unvalidated(height);
+ let block_hash = self.blocks[height].block_hash();
+ block_header.validate(block_hash).unwrap()
+ }
+
+ fn at_height_unvalidated(&self, height: usize) -> BlockHeaderData {
+ assert!(!self.blocks.is_empty());
+ assert!(height < self.blocks.len());
+ BlockHeaderData {
+ chainwork: self.blocks[0].header.work() + Uint256::from_u64(height as u64).unwrap(),
+ height: height as u32,
+ header: self.blocks[height].header.clone(),
+ }
+ }
+
+ pub fn tip(&self) -> ValidatedBlockHeader {
+ assert!(!self.blocks.is_empty());
+ self.at_height(self.blocks.len() - 1)
+ }
+
+ pub fn disconnect_tip(&mut self) -> Option<Block> {
+ self.blocks.pop()
+ }
+
+ pub fn header_cache(&self, heights: std::ops::RangeInclusive<usize>) -> UnboundedCache {
+ let mut cache = UnboundedCache::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
+ }
+}
+
+impl BlockSource for Blockchain {
+ fn get_header<'a>(&'a mut self, header_hash: &'a BlockHash, _height_hint: Option<u32>) -> AsyncBlockSourceResult<'a, BlockHeaderData> {
+ Box::pin(async move {
+ if self.without_headers {
+ return Err(BlockSourceError::persistent("header not found"));
+ }
+
+ for (height, block) in self.blocks.iter().enumerate() {
+ if block.header.block_hash() == *header_hash {
+ let mut header_data = self.at_height_unvalidated(height);
+ if self.malformed_headers {
+ header_data.header.time += 1;
+ }
+
+ return Ok(header_data);
+ }
+ }
+ Err(BlockSourceError::transient("header not found"))
+ })
+ }
+
+ fn get_block<'a>(&'a mut self, header_hash: &'a BlockHash) -> AsyncBlockSourceResult<'a, Block> {
+ Box::pin(async move {
+ for (height, block) in self.blocks.iter().enumerate() {
+ if block.header.block_hash() == *header_hash {
+ if let Some(without_blocks) = &self.without_blocks {
+ if without_blocks.contains(&height) {
+ return Err(BlockSourceError::persistent("block not found"));
+ }
+ }
+
+ return Ok(block.clone());
+ }
+ }
+ Err(BlockSourceError::transient("block not found"))
+ })
+ }
+
+ fn get_best_block<'a>(&'a mut self) -> AsyncBlockSourceResult<'a, (BlockHash, Option<u32>)> {
+ Box::pin(async move {
+ match self.blocks.last() {
+ None => Err(BlockSourceError::transient("empty chain")),
+ Some(block) => {
+ let height = (self.blocks.len() - 1) as u32;
+ Ok((block.block_hash(), Some(height)))
+ },
+ }
+ })
+ }
+}
+
+pub struct NullChainListener;
+
+impl chain::Listen for NullChainListener {
+ fn block_connected(&self, _block: &Block, _height: u32) {}
+ fn block_disconnected(&self, _header: &BlockHeader, _height: u32) {}
+}
+
+pub struct MockChainListener {
+ expected_blocks_connected: RefCell<VecDeque<BlockHeaderData>>,
+ expected_blocks_disconnected: RefCell<VecDeque<BlockHeaderData>>,
+}
+
+impl MockChainListener {
+ pub fn new() -> Self {
+ Self {
+ expected_blocks_connected: RefCell::new(VecDeque::new()),
+ expected_blocks_disconnected: RefCell::new(VecDeque::new()),
+ }
+ }
+
+ pub fn expect_block_connected(self, block: BlockHeaderData) -> Self {
+ self.expected_blocks_connected.borrow_mut().push_back(block);
+ self
+ }
+
+ pub fn expect_block_disconnected(self, block: BlockHeaderData) -> Self {
+ self.expected_blocks_disconnected.borrow_mut().push_back(block);
+ self
+ }
+}
+
+impl chain::Listen for MockChainListener {
+ fn block_connected(&self, block: &Block, height: u32) {
+ match self.expected_blocks_connected.borrow_mut().pop_front() {
+ None => {
+ panic!("Unexpected block connected: {:?}", block.block_hash());
+ },
+ Some(expected_block) => {
+ assert_eq!(block.block_hash(), expected_block.header.block_hash());
+ assert_eq!(height, expected_block.height);
+ },
+ }
+ }
+
+ fn block_disconnected(&self, header: &BlockHeader, height: u32) {
+ match self.expected_blocks_disconnected.borrow_mut().pop_front() {
+ None => {
+ panic!("Unexpected block disconnected: {:?}", header.block_hash());
+ },
+ Some(expected_block) => {
+ assert_eq!(header.block_hash(), expected_block.header.block_hash());
+ assert_eq!(height, expected_block.height);
+ },
+ }
+ }
+}
+
+impl Drop for MockChainListener {
+ fn drop(&mut self) {
+ if std::thread::panicking() {
+ return;
+ }
+
+ let expected_blocks_connected = self.expected_blocks_connected.borrow();
+ if !expected_blocks_connected.is_empty() {
+ panic!("Expected blocks connected: {:?}", expected_blocks_connected);
+ }
+
+ let expected_blocks_disconnected = self.expected_blocks_disconnected.borrow();
+ if !expected_blocks_disconnected.is_empty() {
+ panic!("Expected blocks disconnected: {:?}", expected_blocks_disconnected);
+ }
+ }
+}
//! [`ChannelMonitor`]: ../channelmonitor/struct.ChannelMonitor.html
//! [`MonitorEvent`]: ../channelmonitor/enum.MonitorEvent.html
-use bitcoin::blockdata::block::BlockHeader;
+use bitcoin::blockdata::block::{Block, BlockHeader};
use chain;
use chain::Filter;
}
}
+impl<ChannelSigner: Sign, C: Deref + Send + Sync, T: Deref + Send + Sync, F: Deref + Send + Sync, L: Deref + Send + Sync, P: Deref + Send + Sync>
+chain::Listen for ChainMonitor<ChannelSigner, C, T, F, L, P>
+where
+ ChannelSigner: Sign,
+ C::Target: chain::Filter,
+ T::Target: BroadcasterInterface,
+ F::Target: FeeEstimator,
+ L::Target: Logger,
+ P::Target: channelmonitor::Persist<ChannelSigner>,
+{
+ fn block_connected(&self, block: &Block, height: u32) {
+ let txdata: Vec<_> = block.txdata.iter().enumerate().collect();
+ ChainMonitor::block_connected(self, &block.header, &txdata, height);
+ }
+
+ fn block_disconnected(&self, header: &BlockHeader, height: u32) {
+ ChainMonitor::block_disconnected(self, header, height);
+ }
+}
+
impl<ChannelSigner: Sign, C: Deref + Sync + Send, T: Deref + Sync + Send, F: Deref + Sync + Send, L: Deref + Sync + Send, P: Deref + Sync + Send>
chain::Watch<ChannelSigner> for ChainMonitor<ChannelSigner, C, T, F, L, P>
where C::Target: chain::Filter,
//!
//! [`chain::Watch`]: ../trait.Watch.html
-use bitcoin::blockdata::block::BlockHeader;
+use bitcoin::blockdata::block::{Block, BlockHeader};
use bitcoin::blockdata::transaction::{TxOut,Transaction};
use bitcoin::blockdata::transaction::OutPoint as BitcoinOutPoint;
use bitcoin::blockdata::script::{Script, Builder};
use ln::chan_utils::{CounterpartyCommitmentSecrets, HTLCOutputInCommitment, HTLCType, ChannelTransactionParameters, HolderCommitmentTransaction};
use ln::channelmanager::{HTLCSource, PaymentPreimage, PaymentHash};
use ln::onchaintx::{OnchainTxHandler, InputDescriptors};
+use chain;
use chain::chaininterface::{BroadcasterInterface, FeeEstimator};
use chain::transaction::{OutPoint, TransactionData};
use chain::keysinterface::{SpendableOutputDescriptor, StaticPaymentOutputDescriptor, DelayedPaymentOutputDescriptor, Sign, KeysInterface};
use util::byte_utils;
use util::events::Event;
+use std::cell::RefCell;
use std::collections::{HashMap, HashSet, hash_map};
use std::{cmp, mem};
use std::ops::Deref;
fn update_persisted_channel(&self, id: OutPoint, update: &ChannelMonitorUpdate, data: &ChannelMonitor<ChannelSigner>) -> Result<(), ChannelMonitorUpdateErr>;
}
+impl<Signer: Sign, T: Deref, F: Deref, L: Deref> chain::Listen for (RefCell<ChannelMonitor<Signer>>, T, F, L)
+where
+ T::Target: BroadcasterInterface,
+ F::Target: FeeEstimator,
+ L::Target: Logger,
+{
+ fn block_connected(&self, block: &Block, height: u32) {
+ let txdata: Vec<_> = block.txdata.iter().enumerate().collect();
+ self.0.borrow_mut().block_connected(&block.header, &txdata, height, &*self.1, &*self.2, &*self.3);
+ }
+
+ fn block_disconnected(&self, header: &BlockHeader, height: u32) {
+ self.0.borrow_mut().block_disconnected(header, height, &*self.1, &*self.2, &*self.3);
+ }
+}
+
const MAX_ALLOC_SIZE: usize = 64*1024;
impl<'a, Signer: Sign, K: KeysInterface<Signer = Signer>> ReadableArgs<&'a K>
//! Structs and traits which allow other parts of rust-lightning to interact with the blockchain.
+use bitcoin::blockdata::block::{Block, BlockHeader};
use bitcoin::blockdata::script::Script;
use bitcoin::blockdata::transaction::TxOut;
use bitcoin::hash_types::{BlockHash, Txid};
fn get_utxo(&self, genesis_hash: &BlockHash, short_channel_id: u64) -> Result<TxOut, AccessError>;
}
+/// The `Listen` trait is used to be notified of when blocks have been connected or disconnected
+/// from the chain.
+///
+/// Useful when needing to replay chain data upon startup or as new chain events occur.
+pub trait Listen {
+ /// Notifies the listener that a block was added at the given height.
+ fn block_connected(&self, block: &Block, height: u32);
+
+ /// Notifies the listener that a block was removed at the given height.
+ fn block_disconnected(&self, header: &BlockHeader, height: u32);
+}
+
/// The `Watch` trait defines behavior for watching on-chain activity pertaining to channels as
/// blocks are connected and disconnected.
///
/// `script_pubkey` as the spending condition.
fn register_output(&self, outpoint: &OutPoint, script_pubkey: &Script);
}
+
+impl<T: Listen> Listen for std::ops::Deref<Target = T> {
+ fn block_connected(&self, block: &Block, height: u32) {
+ (**self).block_connected(block, height);
+ }
+
+ fn block_disconnected(&self, header: &BlockHeader, height: u32) {
+ (**self).block_disconnected(header, height);
+ }
+}
+
+impl<T: std::ops::Deref, U: std::ops::Deref> Listen for (T, U)
+where
+ T::Target: Listen,
+ U::Target: Listen,
+{
+ fn block_connected(&self, block: &Block, height: u32) {
+ self.0.block_connected(block, height);
+ self.1.block_connected(block, height);
+ }
+
+ fn block_disconnected(&self, header: &BlockHeader, height: u32) {
+ self.0.block_disconnected(header, height);
+ self.1.block_disconnected(header, height);
+ }
+}
//! imply it needs to fail HTLCs/payments/channels it manages).
//!
-use bitcoin::blockdata::block::BlockHeader;
+use bitcoin::blockdata::block::{Block, BlockHeader};
use bitcoin::blockdata::constants::genesis_block;
use bitcoin::network::constants::Network;
}
}
+impl<Signer: Sign, M: Deref, T: Deref, K: Deref, F: Deref, L: Deref> chain::Listen for ChannelManager<Signer, M, T, K, F, L>
+where
+ M::Target: chain::Watch<Signer>,
+ T::Target: BroadcasterInterface,
+ K::Target: KeysInterface<Signer = Signer>,
+ F::Target: FeeEstimator,
+ L::Target: Logger,
+{
+ fn block_connected(&self, block: &Block, height: u32) {
+ let txdata: Vec<_> = block.txdata.iter().enumerate().collect();
+ ChannelManager::block_connected(self, &block.header, &txdata, height);
+ }
+
+ fn block_disconnected(&self, header: &BlockHeader, _height: u32) {
+ ChannelManager::block_disconnected(self, header);
+ }
+}
+
impl<Signer: Sign, M: Deref, T: Deref, K: Deref, F: Deref, L: Deref> ChannelManager<Signer, M, T, K, F, L>
where M::Target: chain::Watch<Signer>,
T::Target: BroadcasterInterface,
projects / rust-lightning / commitdiff