From b1ecfe705d77b1338c7df5e208d76eb6743f270c Mon Sep 17 00:00:00 2001 From: Jeffrey Czyz Date: Mon, 1 Feb 2021 13:17:20 -0800 Subject: [PATCH] Add ChainNotifier and define ChainListener trait Add an interface for being notified of block connected and disconnected events, along with a notifier for generating such events. Used while polling block sources for a new tip in order to feed these events into ChannelManager and ChainMonitor. --- lightning-block-sync/src/lib.rs | 316 +++++++++++++++++++++++++ lightning-block-sync/src/poll.rs | 4 +- lightning-block-sync/src/test_utils.rs | 103 +++++++- 3 files changed, 419 insertions(+), 4 deletions(-) diff --git a/lightning-block-sync/src/lib.rs b/lightning-block-sync/src/lib.rs index f2716ccc..9228e11f 100644 --- a/lightning-block-sync/src/lib.rs +++ b/lightning-block-sync/src/lib.rs @@ -31,6 +31,8 @@ mod test_utils; #[cfg(any(feature = "rest-client", feature = "rpc-client"))] mod utils; +use crate::poll::{Poll, ValidatedBlockHeader}; + use bitcoin::blockdata::block::{Block, BlockHeader}; use bitcoin::hash_types::BlockHash; use bitcoin::util::uint::Uint256; @@ -130,3 +132,317 @@ pub struct BlockHeaderData { /// of equivalent weight. pub chainwork: Uint256, } + +/// Adaptor used for notifying when blocks have been connected or disconnected from the chain. +/// +/// Used when needing to replay chain data upon startup or as new chain events occur. +pub trait ChainListener { + /// Notifies the listener that a block was added at the given height. + fn block_connected(&mut self, block: &Block, height: u32); + + /// Notifies the listener that a block was removed at the given height. + fn block_disconnected(&mut self, header: &BlockHeader, height: u32); +} + +/// 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; +} + +/// Unbounded cache of block headers keyed by block hash. +pub type UnboundedCache = std::collections::HashMap; + +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 { + self.remove(block_hash) + } +} + +/// Notifies [listeners] of blocks that have been connected or disconnected from the chain. +/// +/// [listeners]: trait.ChainListener.html +struct ChainNotifier { + /// Cache for looking up headers before fetching from a block source. + header_cache: C, +} + +/// 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 { + /// Blocks that were disconnected from the chain since the last poll. + disconnected_blocks: Vec, + + /// Blocks that were connected to the chain since the last poll. + connected_blocks: Vec, +} + +impl ChainNotifier { + /// Finds the fork point 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( + &mut self, + new_header: ValidatedBlockHeader, + old_header: &ValidatedBlockHeader, + chain_poller: &mut P, + chain_listener: &mut L, + ) -> Result<(), (BlockSourceError, Option)> { + let mut difference = self.find_difference(new_header, old_header, chain_poller).await + .map_err(|e| (e, None))?; + + let mut new_tip = *old_header; + for header in difference.disconnected_blocks.drain(..) { + if let Some(cached_header) = self.header_cache.block_disconnected(&header.block_hash) { + assert_eq!(cached_header, header); + } + chain_listener.block_disconnected(&header.header, header.height); + new_tip = header; + } + + for header in difference.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); + chain_listener.block_connected(&block, header.height); + new_tip = header; + } + + Ok(()) + } + + /// 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( + &self, + current_header: ValidatedBlockHeader, + prev_header: &ValidatedBlockHeader, + chain_poller: &mut P, + ) -> BlockSourceResult { + 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?; + } + } + + Ok(ChainDifference { 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( + &self, + chain_poller: &mut P, + header: &ValidatedBlockHeader, + ) -> BlockSourceResult { + 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, + } + } +} + +#[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 mut listener = MockChainListener::new() + .expect_block_connected(*chain.at_height(2)) + .expect_block_connected(*new_tip); + let mut notifier = ChainNotifier { header_cache: chain.header_cache(0..=1) }; + let mut poller = poll::ChainPoller::new(&mut chain, Network::Testnet); + match notifier.synchronize_listener(new_tip, &old_tip, &mut poller, &mut listener).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 mut listener = MockChainListener::new(); + let mut notifier = ChainNotifier { header_cache: main_chain.header_cache(0..=1) }; + let mut poller = poll::ChainPoller::new(&mut test_chain, Network::Testnet); + match notifier.synchronize_listener(new_tip, &old_tip, &mut poller, &mut listener).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 mut listener = MockChainListener::new() + .expect_block_disconnected(*old_tip) + .expect_block_connected(*new_tip); + let mut notifier = ChainNotifier { header_cache: main_chain.header_cache(0..=2) }; + let mut poller = poll::ChainPoller::new(&mut fork_chain, Network::Testnet); + match notifier.synchronize_listener(new_tip, &old_tip, &mut poller, &mut listener).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 mut 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: main_chain.header_cache(0..=3) }; + let mut poller = poll::ChainPoller::new(&mut fork_chain, Network::Testnet); + match notifier.synchronize_listener(new_tip, &old_tip, &mut poller, &mut listener).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 mut 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: main_chain.header_cache(0..=2) }; + let mut poller = poll::ChainPoller::new(&mut fork_chain, Network::Testnet); + match notifier.synchronize_listener(new_tip, &old_tip, &mut poller, &mut listener).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 mut listener = MockChainListener::new(); + let mut notifier = ChainNotifier { header_cache: chain.header_cache(0..=1) }; + let mut poller = poll::ChainPoller::new(&mut chain, Network::Testnet); + match notifier.synchronize_listener(new_tip, &old_tip, &mut poller, &mut listener).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 mut listener = MockChainListener::new(); + let mut notifier = ChainNotifier { header_cache: chain.header_cache(0..=3) }; + let mut poller = poll::ChainPoller::new(&mut chain, Network::Testnet); + match notifier.synchronize_listener(new_tip, &old_tip, &mut poller, &mut listener).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 mut listener = MockChainListener::new() + .expect_block_connected(*chain.at_height(2)); + let mut notifier = ChainNotifier { header_cache: chain.header_cache(0..=3) }; + let mut poller = poll::ChainPoller::new(&mut chain, Network::Testnet); + match notifier.synchronize_listener(new_tip, &old_tip, &mut poller, &mut listener).await { + Err((_, tip)) => assert_eq!(tip, Some(chain.at_height(2))), + Ok(_) => panic!("Expected error"), + } + } +} diff --git a/lightning-block-sync/src/poll.rs b/lightning-block-sync/src/poll.rs index 3ff7606b..34be2437 100644 --- a/lightning-block-sync/src/poll.rs +++ b/lightning-block-sync/src/poll.rs @@ -96,7 +96,7 @@ impl Validate for Block { /// A block header with validated proof of work and corresponding block hash. #[derive(Clone, Copy, Debug, PartialEq)] pub struct ValidatedBlockHeader { - block_hash: BlockHash, + pub(crate) block_hash: BlockHash, inner: BlockHeaderData, } @@ -142,7 +142,7 @@ impl ValidatedBlockHeader { /// A block with validated data against its transaction list and corresponding block hash. pub struct ValidatedBlock { - block_hash: BlockHash, + pub(crate) block_hash: BlockHash, inner: Block, } diff --git a/lightning-block-sync/src/test_utils.rs b/lightning-block-sync/src/test_utils.rs index efd34f74..70a8982a 100644 --- a/lightning-block-sync/src/test_utils.rs +++ b/lightning-block-sync/src/test_utils.rs @@ -1,4 +1,4 @@ -use crate::{AsyncBlockSourceResult, BlockHeaderData, BlockSource, BlockSourceError}; +use crate::{AsyncBlockSourceResult, BlockHeaderData, BlockSource, BlockSourceError, ChainListener, UnboundedCache}; use crate::poll::{Validate, ValidatedBlockHeader}; use bitcoin::blockdata::block::{Block, BlockHeader}; @@ -7,9 +7,12 @@ use bitcoin::hash_types::BlockHash; use bitcoin::network::constants::Network; use bitcoin::util::uint::Uint256; +use std::collections::VecDeque; + #[derive(Default)] pub struct Blockchain { pub blocks: Vec, + without_blocks: Option>, without_headers: bool, malformed_headers: bool, } @@ -46,6 +49,10 @@ impl Blockchain { self } + pub fn without_blocks(self, range: std::ops::RangeFrom) -> Self { + Self { without_blocks: Some(range), ..self } + } + pub fn without_headers(self) -> Self { Self { without_headers: true, ..self } } @@ -54,6 +61,18 @@ impl Blockchain { 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(); @@ -78,6 +97,16 @@ impl Blockchain { pub fn disconnect_tip(&mut self) -> Option { self.blocks.pop() } + + pub fn header_cache(&self, heights: std::ops::RangeInclusive) -> 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 { @@ -103,8 +132,14 @@ impl BlockSource for Blockchain { fn get_block<'a>(&'a mut self, header_hash: &'a BlockHash) -> AsyncBlockSourceResult<'a, Block> { Box::pin(async move { - for block in self.blocks.iter() { + 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()); } } @@ -124,3 +159,67 @@ impl BlockSource for Blockchain { }) } } + +pub struct MockChainListener { + expected_blocks_connected: VecDeque, + expected_blocks_disconnected: VecDeque, +} + +impl MockChainListener { + pub fn new() -> Self { + Self { + expected_blocks_connected: VecDeque::new(), + expected_blocks_disconnected: VecDeque::new(), + } + } + + pub fn expect_block_connected(mut self, block: BlockHeaderData) -> Self { + self.expected_blocks_connected.push_back(block); + self + } + + pub fn expect_block_disconnected(mut self, block: BlockHeaderData) -> Self { + self.expected_blocks_disconnected.push_back(block); + self + } +} + +impl ChainListener for MockChainListener { + fn block_connected(&mut self, block: &Block, height: u32) { + match self.expected_blocks_connected.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(&mut self, header: &BlockHeader, height: u32) { + match self.expected_blocks_disconnected.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; + } + if !self.expected_blocks_connected.is_empty() { + panic!("Expected blocks connected: {:?}", self.expected_blocks_connected); + } + if !self.expected_blocks_disconnected.is_empty() { + panic!("Expected blocks disconnected: {:?}", self.expected_blocks_disconnected); + } + } +} -- 2.30.2