Change ChannelMonitor deserialization to return an optional blockhash.
[rust-lightning] / lightning-block-sync / src / init.rs
1 use crate::{BlockSource, BlockSourceResult, Cache, ChainNotifier};
2 use crate::poll::{ChainPoller, Validate, ValidatedBlockHeader};
3
4 use bitcoin::blockdata::block::{Block, BlockHeader};
5 use bitcoin::hash_types::BlockHash;
6 use bitcoin::network::constants::Network;
7
8 use lightning::chain;
9
10 /// Performs a one-time sync of chain listeners using a single *trusted* block source, bringing each
11 /// listener's view of the chain from its paired block hash to `block_source`'s best chain tip.
12 ///
13 /// Upon success, the returned header can be used to initialize [`SpvClient`]. In the case of
14 /// failure, each listener may be left at a different block hash than the one it was originally
15 /// paired with.
16 ///
17 /// Useful during startup to bring the [`ChannelManager`] and each [`ChannelMonitor`] in sync before
18 /// switching to [`SpvClient`]. For example:
19 ///
20 /// ```
21 /// use bitcoin::hash_types::BlockHash;
22 /// use bitcoin::network::constants::Network;
23 ///
24 /// use lightning::chain;
25 /// use lightning::chain::Watch;
26 /// use lightning::chain::chainmonitor::ChainMonitor;
27 /// use lightning::chain::channelmonitor;
28 /// use lightning::chain::channelmonitor::ChannelMonitor;
29 /// use lightning::chain::chaininterface::BroadcasterInterface;
30 /// use lightning::chain::chaininterface::FeeEstimator;
31 /// use lightning::chain::keysinterface;
32 /// use lightning::chain::keysinterface::KeysInterface;
33 /// use lightning::ln::channelmanager::ChannelManager;
34 /// use lightning::ln::channelmanager::ChannelManagerReadArgs;
35 /// use lightning::util::config::UserConfig;
36 /// use lightning::util::logger::Logger;
37 /// use lightning::util::ser::ReadableArgs;
38 ///
39 /// use lightning_block_sync::*;
40 ///
41 /// use std::io::Cursor;
42 ///
43 /// async fn init_sync<
44 ///     B: BlockSource,
45 ///     K: KeysInterface<Signer = S>,
46 ///     S: keysinterface::Sign,
47 ///     T: BroadcasterInterface,
48 ///     F: FeeEstimator,
49 ///     L: Logger,
50 ///     C: chain::Filter,
51 ///     P: channelmonitor::Persist<S>,
52 /// >(
53 ///     block_source: &mut B,
54 ///     chain_monitor: &ChainMonitor<S, &C, &T, &F, &L, &P>,
55 ///     config: UserConfig,
56 ///     keys_manager: &K,
57 ///     tx_broadcaster: &T,
58 ///     fee_estimator: &F,
59 ///     logger: &L,
60 ///     persister: &P,
61 /// ) {
62 ///     // Read a serialized channel monitor paired with the block hash when it was persisted.
63 ///     let serialized_monitor = "...";
64 ///     let (monitor_block_hash_option, mut monitor) = <(Option<BlockHash>, ChannelMonitor<S>)>::read(
65 ///             &mut Cursor::new(&serialized_monitor), keys_manager).unwrap();
66 ///
67 ///     // Read the channel manager paired with the block hash when it was persisted.
68 ///     let serialized_manager = "...";
69 ///     let (manager_block_hash_option, mut manager) = {
70 ///             let read_args = ChannelManagerReadArgs::new(
71 ///                     keys_manager,
72 ///                     fee_estimator,
73 ///                     chain_monitor,
74 ///                     tx_broadcaster,
75 ///                     logger,
76 ///                     config,
77 ///                     vec![&mut monitor],
78 ///             );
79 ///             <(Option<BlockHash>, ChannelManager<S, &ChainMonitor<S, &C, &T, &F, &L, &P>, &T, &K, &F, &L>)>::read(
80 ///                     &mut Cursor::new(&serialized_manager), read_args).unwrap()
81 ///     };
82 ///
83 ///     // Synchronize any channel monitors and the channel manager to be on the best block.
84 ///     let mut cache = UnboundedCache::new();
85 ///     let mut monitor_listener = (monitor, &*tx_broadcaster, &*fee_estimator, &*logger);
86 ///     let mut listeners = vec![];
87 ///     if let Some(monitor_block_hash) = monitor_block_hash_option {
88 ///             listeners.push((monitor_block_hash, &mut monitor_listener as &mut dyn chain::Listen))
89 ///     }
90 ///     if let Some(manager_block_hash) = manager_block_hash_option {
91 ///             listeners.push((manager_block_hash, &mut manager as &mut dyn chain::Listen))
92 ///     }
93 ///     let chain_tip = init::synchronize_listeners(
94 ///             block_source, Network::Bitcoin, &mut cache, listeners).await.unwrap();
95 ///
96 ///     // Allow the chain monitor to watch any channels.
97 ///     let monitor = monitor_listener.0;
98 ///     chain_monitor.watch_channel(monitor.get_funding_txo().0, monitor);
99 ///
100 ///     // Create an SPV client to notify the chain monitor and channel manager of block events.
101 ///     let chain_poller = poll::ChainPoller::new(block_source, Network::Bitcoin);
102 ///     let mut chain_listener = (chain_monitor, &manager);
103 ///     let spv_client = SpvClient::new(chain_tip, chain_poller, &mut cache, &chain_listener);
104 /// }
105 /// ```
106 ///
107 /// [`SpvClient`]: ../struct.SpvClient.html
108 /// [`ChannelManager`]: ../../lightning/ln/channelmanager/struct.ChannelManager.html
109 /// [`ChannelMonitor`]: ../../lightning/chain/channelmonitor/struct.ChannelMonitor.html
110 pub async fn synchronize_listeners<B: BlockSource, C: Cache>(
111         block_source: &mut B,
112         network: Network,
113         header_cache: &mut C,
114         mut chain_listeners: Vec<(BlockHash, &mut dyn chain::Listen)>,
115 ) -> BlockSourceResult<ValidatedBlockHeader> {
116         let (best_block_hash, best_block_height) = block_source.get_best_block().await?;
117         let best_header = block_source
118                 .get_header(&best_block_hash, best_block_height).await?
119                 .validate(best_block_hash)?;
120
121         // Fetch the header for the block hash paired with each listener.
122         let mut chain_listeners_with_old_headers = Vec::new();
123         for (old_block_hash, chain_listener) in chain_listeners.drain(..) {
124                 let old_header = match header_cache.look_up(&old_block_hash) {
125                         Some(header) => *header,
126                         None => block_source
127                                 .get_header(&old_block_hash, None).await?
128                                 .validate(old_block_hash)?
129                 };
130                 chain_listeners_with_old_headers.push((old_header, chain_listener))
131         }
132
133         // Find differences and disconnect blocks for each listener individually.
134         let mut chain_poller = ChainPoller::new(block_source, network);
135         let mut chain_listeners_at_height = Vec::new();
136         let mut most_common_ancestor = None;
137         let mut most_connected_blocks = Vec::new();
138         for (old_header, chain_listener) in chain_listeners_with_old_headers.drain(..) {
139                 // Disconnect any stale blocks, but keep them in the cache for the next iteration.
140                 let header_cache = &mut ReadOnlyCache(header_cache);
141                 let (common_ancestor, connected_blocks) = {
142                         let chain_listener = &DynamicChainListener(chain_listener);
143                         let mut chain_notifier = ChainNotifier { header_cache, chain_listener };
144                         let difference =
145                                 chain_notifier.find_difference(best_header, &old_header, &mut chain_poller).await?;
146                         chain_notifier.disconnect_blocks(difference.disconnected_blocks);
147                         (difference.common_ancestor, difference.connected_blocks)
148                 };
149
150                 // Keep track of the most common ancestor and all blocks connected across all listeners.
151                 chain_listeners_at_height.push((common_ancestor.height, chain_listener));
152                 if connected_blocks.len() > most_connected_blocks.len() {
153                         most_common_ancestor = Some(common_ancestor);
154                         most_connected_blocks = connected_blocks;
155                 }
156         }
157
158         // Connect new blocks for all listeners at once to avoid re-fetching blocks.
159         if let Some(common_ancestor) = most_common_ancestor {
160                 let chain_listener = &ChainListenerSet(chain_listeners_at_height);
161                 let mut chain_notifier = ChainNotifier { header_cache, chain_listener };
162                 chain_notifier.connect_blocks(common_ancestor, most_connected_blocks, &mut chain_poller)
163                         .await.or_else(|(e, _)| Err(e))?;
164         }
165
166         Ok(best_header)
167 }
168
169 /// A wrapper to make a cache read-only.
170 ///
171 /// Used to prevent losing headers that may be needed to disconnect blocks common to more than one
172 /// listener.
173 struct ReadOnlyCache<'a, C: Cache>(&'a mut C);
174
175 impl<'a, C: Cache> Cache for ReadOnlyCache<'a, C> {
176         fn look_up(&self, block_hash: &BlockHash) -> Option<&ValidatedBlockHeader> {
177                 self.0.look_up(block_hash)
178         }
179
180         fn block_connected(&mut self, _block_hash: BlockHash, _block_header: ValidatedBlockHeader) {
181                 unreachable!()
182         }
183
184         fn block_disconnected(&mut self, _block_hash: &BlockHash) -> Option<ValidatedBlockHeader> {
185                 None
186         }
187 }
188
189 /// Wrapper for supporting dynamically sized chain listeners.
190 struct DynamicChainListener<'a>(&'a mut dyn chain::Listen);
191
192 impl<'a> chain::Listen for DynamicChainListener<'a> {
193         fn block_connected(&self, _block: &Block, _height: u32) {
194                 unreachable!()
195         }
196
197         fn block_disconnected(&self, header: &BlockHeader, height: u32) {
198                 self.0.block_disconnected(header, height)
199         }
200 }
201
202 /// A set of dynamically sized chain listeners, each paired with a starting block height.
203 struct ChainListenerSet<'a>(Vec<(u32, &'a mut dyn chain::Listen)>);
204
205 impl<'a> chain::Listen for ChainListenerSet<'a> {
206         fn block_connected(&self, block: &Block, height: u32) {
207                 for (starting_height, chain_listener) in self.0.iter() {
208                         if height > *starting_height {
209                                 chain_listener.block_connected(block, height);
210                         }
211                 }
212         }
213
214         fn block_disconnected(&self, _header: &BlockHeader, _height: u32) {
215                 unreachable!()
216         }
217 }
218
219 #[cfg(test)]
220 mod tests {
221         use crate::test_utils::{Blockchain, MockChainListener};
222         use super::*;
223
224         use bitcoin::network::constants::Network;
225
226         #[tokio::test]
227         async fn sync_from_same_chain() {
228                 let mut chain = Blockchain::default().with_height(4);
229
230                 let mut listener_1 = MockChainListener::new()
231                         .expect_block_connected(*chain.at_height(2))
232                         .expect_block_connected(*chain.at_height(3))
233                         .expect_block_connected(*chain.at_height(4));
234                 let mut listener_2 = MockChainListener::new()
235                         .expect_block_connected(*chain.at_height(3))
236                         .expect_block_connected(*chain.at_height(4));
237                 let mut listener_3 = MockChainListener::new()
238                         .expect_block_connected(*chain.at_height(4));
239
240                 let listeners = vec![
241                         (chain.at_height(1).block_hash, &mut listener_1 as &mut dyn chain::Listen),
242                         (chain.at_height(2).block_hash, &mut listener_2 as &mut dyn chain::Listen),
243                         (chain.at_height(3).block_hash, &mut listener_3 as &mut dyn chain::Listen),
244                 ];
245                 let mut cache = chain.header_cache(0..=4);
246                 match synchronize_listeners(&mut chain, Network::Bitcoin, &mut cache, listeners).await {
247                         Ok(header) => assert_eq!(header, chain.tip()),
248                         Err(e) => panic!("Unexpected error: {:?}", e),
249                 }
250         }
251
252         #[tokio::test]
253         async fn sync_from_different_chains() {
254                 let mut main_chain = Blockchain::default().with_height(4);
255                 let fork_chain_1 = main_chain.fork_at_height(1);
256                 let fork_chain_2 = main_chain.fork_at_height(2);
257                 let fork_chain_3 = main_chain.fork_at_height(3);
258
259                 let mut listener_1 = MockChainListener::new()
260                         .expect_block_disconnected(*fork_chain_1.at_height(4))
261                         .expect_block_disconnected(*fork_chain_1.at_height(3))
262                         .expect_block_disconnected(*fork_chain_1.at_height(2))
263                         .expect_block_connected(*main_chain.at_height(2))
264                         .expect_block_connected(*main_chain.at_height(3))
265                         .expect_block_connected(*main_chain.at_height(4));
266                 let mut listener_2 = MockChainListener::new()
267                         .expect_block_disconnected(*fork_chain_2.at_height(4))
268                         .expect_block_disconnected(*fork_chain_2.at_height(3))
269                         .expect_block_connected(*main_chain.at_height(3))
270                         .expect_block_connected(*main_chain.at_height(4));
271                 let mut listener_3 = MockChainListener::new()
272                         .expect_block_disconnected(*fork_chain_3.at_height(4))
273                         .expect_block_connected(*main_chain.at_height(4));
274
275                 let listeners = vec![
276                         (fork_chain_1.tip().block_hash, &mut listener_1 as &mut dyn chain::Listen),
277                         (fork_chain_2.tip().block_hash, &mut listener_2 as &mut dyn chain::Listen),
278                         (fork_chain_3.tip().block_hash, &mut listener_3 as &mut dyn chain::Listen),
279                 ];
280                 let mut cache = fork_chain_1.header_cache(2..=4);
281                 cache.extend(fork_chain_2.header_cache(3..=4));
282                 cache.extend(fork_chain_3.header_cache(4..=4));
283                 match synchronize_listeners(&mut main_chain, Network::Bitcoin, &mut cache, listeners).await {
284                         Ok(header) => assert_eq!(header, main_chain.tip()),
285                         Err(e) => panic!("Unexpected error: {:?}", e),
286                 }
287         }
288
289         #[tokio::test]
290         async fn sync_from_overlapping_chains() {
291                 let mut main_chain = Blockchain::default().with_height(4);
292                 let fork_chain_1 = main_chain.fork_at_height(1);
293                 let fork_chain_2 = fork_chain_1.fork_at_height(2);
294                 let fork_chain_3 = fork_chain_2.fork_at_height(3);
295
296                 let mut listener_1 = MockChainListener::new()
297                         .expect_block_disconnected(*fork_chain_1.at_height(4))
298                         .expect_block_disconnected(*fork_chain_1.at_height(3))
299                         .expect_block_disconnected(*fork_chain_1.at_height(2))
300                         .expect_block_connected(*main_chain.at_height(2))
301                         .expect_block_connected(*main_chain.at_height(3))
302                         .expect_block_connected(*main_chain.at_height(4));
303                 let mut listener_2 = MockChainListener::new()
304                         .expect_block_disconnected(*fork_chain_2.at_height(4))
305                         .expect_block_disconnected(*fork_chain_2.at_height(3))
306                         .expect_block_disconnected(*fork_chain_2.at_height(2))
307                         .expect_block_connected(*main_chain.at_height(2))
308                         .expect_block_connected(*main_chain.at_height(3))
309                         .expect_block_connected(*main_chain.at_height(4));
310                 let mut listener_3 = MockChainListener::new()
311                         .expect_block_disconnected(*fork_chain_3.at_height(4))
312                         .expect_block_disconnected(*fork_chain_3.at_height(3))
313                         .expect_block_disconnected(*fork_chain_3.at_height(2))
314                         .expect_block_connected(*main_chain.at_height(2))
315                         .expect_block_connected(*main_chain.at_height(3))
316                         .expect_block_connected(*main_chain.at_height(4));
317
318                 let listeners = vec![
319                         (fork_chain_1.tip().block_hash, &mut listener_1 as &mut dyn chain::Listen),
320                         (fork_chain_2.tip().block_hash, &mut listener_2 as &mut dyn chain::Listen),
321                         (fork_chain_3.tip().block_hash, &mut listener_3 as &mut dyn chain::Listen),
322                 ];
323                 let mut cache = fork_chain_1.header_cache(2..=4);
324                 cache.extend(fork_chain_2.header_cache(3..=4));
325                 cache.extend(fork_chain_3.header_cache(4..=4));
326                 match synchronize_listeners(&mut main_chain, Network::Bitcoin, &mut cache, listeners).await {
327                         Ok(header) => assert_eq!(header, main_chain.tip()),
328                         Err(e) => panic!("Unexpected error: {:?}", e),
329                 }
330         }
331
332         #[tokio::test]
333         async fn cache_connected_and_keep_disconnected_blocks() {
334                 let mut main_chain = Blockchain::default().with_height(2);
335                 let fork_chain = main_chain.fork_at_height(1);
336                 let new_tip = main_chain.tip();
337                 let old_tip = fork_chain.tip();
338
339                 let mut listener = MockChainListener::new()
340                         .expect_block_disconnected(*old_tip)
341                         .expect_block_connected(*new_tip);
342
343                 let listeners = vec![(old_tip.block_hash, &mut listener as &mut dyn chain::Listen)];
344                 let mut cache = fork_chain.header_cache(2..=2);
345                 match synchronize_listeners(&mut main_chain, Network::Bitcoin, &mut cache, listeners).await {
346                         Ok(_) => {
347                                 assert!(cache.contains_key(&new_tip.block_hash));
348                                 assert!(cache.contains_key(&old_tip.block_hash));
349                         },
350                         Err(e) => panic!("Unexpected error: {:?}", e),
351                 }
352         }
353 }