]> git.bitcoin.ninja Git - rust-lightning/commitdiff
Utility for syncing a set of chain listeners
authorJeffrey Czyz <jkczyz@gmail.com>
Fri, 5 Feb 2021 03:20:03 +0000 (19:20 -0800)
committerJeffrey Czyz <jkczyz@gmail.com>
Fri, 26 Feb 2021 06:54:43 +0000 (00:54 -0600)
Add a utility for syncing a set of chain listeners to a common chain
tip. Required to use before creating an SpvClient when the chain
listener used with the client is actually a set of listeners each of
which may have had left off at a different block. This would occur when
the listeners had been persisted individually at different frequencies
(e.g., a ChainMonitor's individual ChannelMonitors).

lightning-block-sync/src/init.rs [new file with mode: 0644]
lightning-block-sync/src/lib.rs
lightning-block-sync/src/test_utils.rs
lightning/src/chain/chainmonitor.rs
lightning/src/chain/channelmonitor.rs
lightning/src/chain/mod.rs
lightning/src/ln/channelmanager.rs

diff --git a/lightning-block-sync/src/init.rs b/lightning-block-sync/src/init.rs
new file mode 100644 (file)
index 0000000..da9895a
--- /dev/null
@@ -0,0 +1,351 @@
+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),
+               }
+       }
+}
index 5ed8689fb965e00ee1d7305e790207d339d2c9c0..db536fe7d43778ccb062be5813548f2072353a44 100644 (file)
@@ -19,6 +19,7 @@
 #[cfg(any(feature = "rest-client", feature = "rpc-client"))]
 pub mod http;
 
+pub mod init;
 pub mod poll;
 
 #[cfg(feature = "rest-client")]
@@ -42,7 +43,11 @@ 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.
@@ -154,22 +159,11 @@ pub struct BlockHeaderData {
 /// 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<P: Poll, C: Cache, L: ChainListener> {
+pub struct SpvClient<'a, P: Poll, C: Cache, L: Deref>
+where L::Target: chain::Listen {
        chain_tip: ValidatedBlockHeader,
        chain_poller: P,
-       chain_notifier: ChainNotifier<C>,
-       chain_listener: L,
-}
-
-/// 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);
+       chain_notifier: ChainNotifier<'a, C, L>,
 }
 
 /// The `Cache` trait defines behavior for managing a block header cache, where block headers are
@@ -214,7 +208,7 @@ impl Cache for UnboundedCache {
        }
 }
 
-impl<P: Poll, C: Cache, L: ChainListener> SpvClient<P, C, L> {
+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
@@ -228,11 +222,11 @@ impl<P: Poll, C: Cache, L: ChainListener> SpvClient<P, C, L> {
        pub fn new(
                chain_tip: ValidatedBlockHeader,
                chain_poller: P,
-               header_cache: C,
+               header_cache: &'a mut C,
                chain_listener: L,
        ) -> Self {
-               let chain_notifier = ChainNotifier { header_cache };
-               Self { chain_tip, chain_poller, chain_notifier, chain_listener }
+               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
@@ -262,7 +256,7 @@ impl<P: Poll, C: Cache, L: ChainListener> SpvClient<P, C, L> {
        /// 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, &mut self.chain_listener).await
+                       best_chain_tip, &self.chain_tip, &mut self.chain_poller).await
                {
                        Ok(_) => {
                                self.chain_tip = best_chain_tip;
@@ -279,10 +273,13 @@ impl<P: Poll, C: Cache, L: ChainListener> SpvClient<P, C, L> {
 
 /// Notifies [listeners] of blocks that have been connected or disconnected from the chain.
 ///
-/// [listeners]: trait.ChainListener.html
-struct ChainNotifier<C: Cache> {
+/// [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: C,
+       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
@@ -291,6 +288,11 @@ struct ChainNotifier<C: Cache> {
 /// 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>,
 
@@ -298,45 +300,28 @@ struct ChainDifference {
        connected_blocks: Vec<ValidatedBlockHeader>,
 }
 
-impl<C: Cache> ChainNotifier<C> {
-       /// 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`.
+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<L: ChainListener, P: Poll>(
+       async fn synchronize_listener<P: Poll>(
                &mut self,
                new_header: ValidatedBlockHeader,
                old_header: &ValidatedBlockHeader,
                chain_poller: &mut P,
-               chain_listener: &mut L,
        ) -> Result<(), (BlockSourceError, Option<ValidatedBlockHeader>)> {
-               let mut difference = self.find_difference(new_header, old_header, chain_poller).await
+               let 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(())
+               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
@@ -373,7 +358,8 @@ impl<C: Cache> ChainNotifier<C> {
                        }
                }
 
-               Ok(ChainDifference { disconnected_blocks, connected_blocks })
+               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
@@ -388,6 +374,37 @@ impl<C: Cache> ChainNotifier<C> {
                        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)]
@@ -403,8 +420,9 @@ mod spv_client_tests {
                let best_tip = chain.at_height(1);
 
                let poller = poll::ChainPoller::new(&mut chain, Network::Testnet);
-               let cache = UnboundedCache::new();
-               let mut client = SpvClient::new(best_tip, poller, cache, NullChainListener {});
+               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);
@@ -421,8 +439,9 @@ mod spv_client_tests {
                let common_tip = chain.tip();
 
                let poller = poll::ChainPoller::new(&mut chain, Network::Testnet);
-               let cache = UnboundedCache::new();
-               let mut client = SpvClient::new(common_tip, poller, cache, NullChainListener {});
+               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)) => {
@@ -440,8 +459,9 @@ mod spv_client_tests {
                let old_tip = chain.at_height(1);
 
                let poller = poll::ChainPoller::new(&mut chain, Network::Testnet);
-               let cache = UnboundedCache::new();
-               let mut client = SpvClient::new(old_tip, poller, cache, NullChainListener {});
+               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)) => {
@@ -459,8 +479,9 @@ mod spv_client_tests {
                let old_tip = chain.at_height(1);
 
                let poller = poll::ChainPoller::new(&mut chain, Network::Testnet);
-               let cache = UnboundedCache::new();
-               let mut client = SpvClient::new(old_tip, poller, cache, NullChainListener {});
+               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)) => {
@@ -478,8 +499,9 @@ mod spv_client_tests {
                let old_tip = chain.at_height(1);
 
                let poller = poll::ChainPoller::new(&mut chain, Network::Testnet);
-               let cache = UnboundedCache::new();
-               let mut client = SpvClient::new(old_tip, poller, cache, NullChainListener {});
+               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)) => {
@@ -498,8 +520,9 @@ mod spv_client_tests {
                let worse_tip = chain.tip();
 
                let poller = poll::ChainPoller::new(&mut chain, Network::Testnet);
-               let cache = UnboundedCache::new();
-               let mut client = SpvClient::new(best_tip, poller, cache, NullChainListener {});
+               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)) => {
@@ -524,12 +547,15 @@ mod chain_notifier_tests {
 
                let new_tip = chain.tip();
                let old_tip = chain.at_height(1);
-               let mut listener = MockChainListener::new()
+               let chain_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 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, &mut listener).await {
+               match notifier.synchronize_listener(new_tip, &old_tip, &mut poller).await {
                        Err((e, _)) => panic!("Unexpected error: {:?}", e),
                        Ok(_) => {},
                }
@@ -542,10 +568,13 @@ mod chain_notifier_tests {
 
                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 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, &mut listener).await {
+               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");
@@ -561,12 +590,15 @@ mod chain_notifier_tests {
 
                let new_tip = fork_chain.tip();
                let old_tip = main_chain.tip();
-               let mut listener = MockChainListener::new()
+               let chain_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 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, &mut listener).await {
+               match notifier.synchronize_listener(new_tip, &old_tip, &mut poller).await {
                        Err((e, _)) => panic!("Unexpected error: {:?}", e),
                        Ok(_) => {},
                }
@@ -580,13 +612,16 @@ mod chain_notifier_tests {
 
                let new_tip = fork_chain.tip();
                let old_tip = main_chain.tip();
-               let mut listener = MockChainListener::new()
+               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: main_chain.header_cache(0..=3) };
+               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, &mut listener).await {
+               match notifier.synchronize_listener(new_tip, &old_tip, &mut poller).await {
                        Err((e, _)) => panic!("Unexpected error: {:?}", e),
                        Ok(_) => {},
                }
@@ -600,13 +635,16 @@ mod chain_notifier_tests {
 
                let new_tip = fork_chain.tip();
                let old_tip = main_chain.tip();
-               let mut listener = MockChainListener::new()
+               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: main_chain.header_cache(0..=2) };
+               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, &mut listener).await {
+               match notifier.synchronize_listener(new_tip, &old_tip, &mut poller).await {
                        Err((e, _)) => panic!("Unexpected error: {:?}", e),
                        Ok(_) => {},
                }
@@ -618,10 +656,13 @@ mod chain_notifier_tests {
 
                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 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, &mut listener).await {
+               match notifier.synchronize_listener(new_tip, &old_tip, &mut poller).await {
                        Err((_, tip)) => assert_eq!(tip, None),
                        Ok(_) => panic!("Expected error"),
                }
@@ -633,10 +674,13 @@ mod chain_notifier_tests {
 
                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 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, &mut listener).await {
+               match notifier.synchronize_listener(new_tip, &old_tip, &mut poller).await {
                        Err((_, tip)) => assert_eq!(tip, Some(old_tip)),
                        Ok(_) => panic!("Expected error"),
                }
@@ -648,11 +692,14 @@ mod chain_notifier_tests {
 
                let new_tip = chain.tip();
                let old_tip = chain.at_height(1);
-               let mut listener = MockChainListener::new()
+               let chain_listener = &MockChainListener::new()
                        .expect_block_connected(*chain.at_height(2));
-               let mut notifier = ChainNotifier { header_cache: chain.header_cache(0..=3) };
+               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, &mut listener).await {
+               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"),
                }
index 807a33a69dec3fb305fb56ad9b9dc6953a26b138..8c37c94d8ec4c2146d978170fd66ccf1fae75172 100644 (file)
@@ -1,4 +1,4 @@
-use crate::{AsyncBlockSourceResult, BlockHeaderData, BlockSource, BlockSourceError, ChainListener, UnboundedCache};
+use crate::{AsyncBlockSourceResult, BlockHeaderData, BlockSource, BlockSourceError, UnboundedCache};
 use crate::poll::{Validate, ValidatedBlockHeader};
 
 use bitcoin::blockdata::block::{Block, BlockHeader};
@@ -7,6 +7,9 @@ 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)]
@@ -162,38 +165,38 @@ impl BlockSource for Blockchain {
 
 pub struct NullChainListener;
 
-impl ChainListener for NullChainListener {
-       fn block_connected(&mut self, _block: &Block, _height: u32) {}
-       fn block_disconnected(&mut self, _header: &BlockHeader, _height: u32) {}
+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: VecDeque<BlockHeaderData>,
-       expected_blocks_disconnected: VecDeque<BlockHeaderData>,
+       expected_blocks_connected: RefCell<VecDeque<BlockHeaderData>>,
+       expected_blocks_disconnected: RefCell<VecDeque<BlockHeaderData>>,
 }
 
 impl MockChainListener {
        pub fn new() -> Self {
                Self {
-                       expected_blocks_connected: VecDeque::new(),
-                       expected_blocks_disconnected: VecDeque::new(),
+                       expected_blocks_connected: RefCell::new(VecDeque::new()),
+                       expected_blocks_disconnected: RefCell::new(VecDeque::new()),
                }
        }
 
-       pub fn expect_block_connected(mut self, block: BlockHeaderData) -> Self {
-               self.expected_blocks_connected.push_back(block);
+       pub fn expect_block_connected(self, block: BlockHeaderData) -> Self {
+               self.expected_blocks_connected.borrow_mut().push_back(block);
                self
        }
 
-       pub fn expect_block_disconnected(mut self, block: BlockHeaderData) -> Self {
-               self.expected_blocks_disconnected.push_back(block);
+       pub fn expect_block_disconnected(self, block: BlockHeaderData) -> Self {
+               self.expected_blocks_disconnected.borrow_mut().push_back(block);
                self
        }
 }
 
-impl ChainListener for MockChainListener {
-       fn block_connected(&mut self, block: &Block, height: u32) {
-               match self.expected_blocks_connected.pop_front() {
+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());
                        },
@@ -204,8 +207,8 @@ impl ChainListener for MockChainListener {
                }
        }
 
-       fn block_disconnected(&mut self, header: &BlockHeader, height: u32) {
-               match self.expected_blocks_disconnected.pop_front() {
+       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());
                        },
@@ -222,11 +225,15 @@ impl Drop for MockChainListener {
                if std::thread::panicking() {
                        return;
                }
-               if !self.expected_blocks_connected.is_empty() {
-                       panic!("Expected blocks connected: {:?}", self.expected_blocks_connected);
+
+               let expected_blocks_connected = self.expected_blocks_connected.borrow();
+               if !expected_blocks_connected.is_empty() {
+                       panic!("Expected blocks connected: {:?}", expected_blocks_connected);
                }
-               if !self.expected_blocks_disconnected.is_empty() {
-                       panic!("Expected blocks disconnected: {:?}", self.expected_blocks_disconnected);
+
+               let expected_blocks_disconnected = self.expected_blocks_disconnected.borrow();
+               if !expected_blocks_disconnected.is_empty() {
+                       panic!("Expected blocks disconnected: {:?}", expected_blocks_disconnected);
                }
        }
 }
index 8483f5ca829ff3e1ad5aeecb11b5c97ee1a3a347..4cb9128d92fdbd826bf92f2cd62c5a25cd401dff 100644 (file)
@@ -29,7 +29,7 @@
 //! [`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;
@@ -140,6 +140,26 @@ where C::Target: 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,
index 0266f31e204e2ceaf508f01e6db5f16125a0c6ad..71490dd6e0e1af4de7b19b71e8699da47d5c2f46 100644 (file)
@@ -22,7 +22,7 @@
 //!
 //! [`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};
@@ -41,6 +41,7 @@ use ln::chan_utils;
 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};
@@ -49,6 +50,7 @@ use util::ser::{Readable, ReadableArgs, MaybeReadable, Writer, Writeable, U48};
 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;
@@ -2297,6 +2299,22 @@ pub trait Persist<ChannelSigner: Sign>: Send + Sync {
        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>
index 1d51f262216d3620fb068908864db0c545f0f734..c1fceec8c47171969397558db0ce62d11faf93ed 100644 (file)
@@ -9,6 +9,7 @@
 
 //! 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};
@@ -46,6 +47,18 @@ pub trait Access: Send + Sync {
        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.
 ///
@@ -123,3 +136,29 @@ pub trait Filter: Send + Sync {
        /// `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);
+       }
+}
index a759443d86ed9ab56374c3d44d24ad8affa6a5a9..8f7fbde5f9cb131312ea2ae0d8be6df4b634474a 100644 (file)
@@ -18,7 +18,7 @@
 //! 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;
 
@@ -3139,6 +3139,24 @@ impl<Signer: Sign, M: Deref, T: Deref, K: Deref, F: Deref, L: Deref> EventsProvi
        }
 }
 
+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,