22//! from disk.
33
44use crate :: poll:: { ChainPoller , Validate , ValidatedBlockHeader } ;
5- use crate :: { BlockSource , BlockSourceResult , Cache , ChainNotifier } ;
5+ use crate :: { BlockSource , BlockSourceResult , Cache , ChainNotifier , UnboundedCache } ;
66
77use bitcoin:: block:: Header ;
88use bitcoin:: hash_types:: BlockHash ;
3232/// Performs a one-time sync of chain listeners using a single *trusted* block source, bringing each
3333/// listener's view of the chain from its paired block hash to `block_source`'s best chain tip.
3434///
35- /// Upon success, the returned header can be used to initialize [`SpvClient`]. In the case of
36- /// failure, each listener may be left at a different block hash than the one it was originally
37- /// paired with.
35+ /// Upon success, the returned header and header cache can be used to initialize [`SpvClient`]. In
36+ /// the case of failure, each listener may be left at a different block hash than the one it was
37+ /// originally paired with.
3838///
3939/// Useful during startup to bring the [`ChannelManager`] and each [`ChannelMonitor`] in sync before
4040/// switching to [`SpvClient`]. For example:
@@ -114,14 +114,13 @@ where
114114/// };
115115///
116116/// // Synchronize any channel monitors and the channel manager to be on the best block.
117- /// let mut cache = UnboundedCache::new();
118117/// let mut monitor_listener = (monitor, &*tx_broadcaster, &*fee_estimator, &*logger);
119118/// let listeners = vec![
120119/// (monitor_best_block, &monitor_listener as &dyn chain::Listen),
121120/// (manager_best_block, &manager as &dyn chain::Listen),
122121/// ];
123- /// let chain_tip = init::synchronize_listeners(
124- /// block_source, Network::Bitcoin, &mut cache, listeners).await.unwrap();
122+ /// let (chain_cache, chain_tip) = init::synchronize_listeners(
123+ /// block_source, Network::Bitcoin, listeners).await.unwrap();
125124///
126125/// // Allow the chain monitor to watch any channels.
127126/// let monitor = monitor_listener.0;
@@ -130,21 +129,16 @@ where
130129/// // Create an SPV client to notify the chain monitor and channel manager of block events.
131130/// let chain_poller = poll::ChainPoller::new(block_source, Network::Bitcoin);
132131/// let mut chain_listener = (chain_monitor, &manager);
133- /// let spv_client = SpvClient::new(chain_tip, chain_poller, &mut cache , &chain_listener);
132+ /// let spv_client = SpvClient::new(chain_tip, chain_poller, chain_cache , &chain_listener);
134133/// }
135134/// ```
136135///
137136/// [`SpvClient`]: crate::SpvClient
138137/// [`ChannelManager`]: lightning::ln::channelmanager::ChannelManager
139138/// [`ChannelMonitor`]: lightning::chain::channelmonitor::ChannelMonitor
140- pub async fn synchronize_listeners <
141- B : Deref + Sized + Send + Sync ,
142- C : Cache ,
143- L : chain:: Listen + ?Sized ,
144- > (
145- block_source : B , network : Network , header_cache : & mut C ,
146- mut chain_listeners : Vec < ( BestBlock , & L ) > ,
147- ) -> BlockSourceResult < ValidatedBlockHeader >
139+ pub async fn synchronize_listeners < B : Deref + Sized + Send + Sync , L : chain:: Listen + ?Sized > (
140+ block_source : B , network : Network , mut chain_listeners : Vec < ( BestBlock , & L ) > ,
141+ ) -> BlockSourceResult < ( UnboundedCache , ValidatedBlockHeader ) >
148142where
149143 B :: Target : BlockSource ,
150144{
@@ -155,12 +149,13 @@ where
155149 let mut chain_listeners_at_height = Vec :: new ( ) ;
156150 let mut most_common_ancestor = None ;
157151 let mut most_connected_blocks = Vec :: new ( ) ;
152+ let mut header_cache = UnboundedCache :: new ( ) ;
158153 for ( old_best_block, chain_listener) in chain_listeners. drain ( ..) {
159154 // Disconnect any stale blocks, but keep them in the cache for the next iteration.
160- let header_cache = & mut ReadOnlyCache ( header_cache) ;
161155 let ( common_ancestor, connected_blocks) = {
162156 let chain_listener = & DynamicChainListener ( chain_listener) ;
163- let mut chain_notifier = ChainNotifier { header_cache, chain_listener } ;
157+ let mut chain_notifier =
158+ ChainNotifier { header_cache : & mut header_cache, chain_listener } ;
164159 let difference = chain_notifier
165160 . find_difference_from_best_block ( best_header, old_best_block, & mut chain_poller)
166161 . await ?;
@@ -181,32 +176,14 @@ where
181176 // Connect new blocks for all listeners at once to avoid re-fetching blocks.
182177 if let Some ( common_ancestor) = most_common_ancestor {
183178 let chain_listener = & ChainListenerSet ( chain_listeners_at_height) ;
184- let mut chain_notifier = ChainNotifier { header_cache, chain_listener } ;
179+ let mut chain_notifier = ChainNotifier { header_cache : & mut header_cache , chain_listener } ;
185180 chain_notifier
186181 . connect_blocks ( common_ancestor, most_connected_blocks, & mut chain_poller)
187182 . await
188183 . map_err ( |( e, _) | e) ?;
189184 }
190185
191- Ok ( best_header)
192- }
193-
194- /// A wrapper to make a cache read-only.
195- ///
196- /// Used to prevent losing headers that may be needed to disconnect blocks common to more than one
197- /// listener.
198- struct ReadOnlyCache < ' a , C : Cache > ( & ' a mut C ) ;
199-
200- impl < ' a , C : Cache > Cache for ReadOnlyCache < ' a , C > {
201- fn look_up ( & self , block_hash : & BlockHash ) -> Option < & ValidatedBlockHeader > {
202- self . 0 . look_up ( block_hash)
203- }
204-
205- fn block_connected ( & mut self , _block_hash : BlockHash , _block_header : ValidatedBlockHeader ) {
206- unreachable ! ( )
207- }
208-
209- fn blocks_disconnected ( & mut self , _fork_point : & ValidatedBlockHeader ) { }
186+ Ok ( ( header_cache, best_header) )
210187}
211188
212189/// Wrapper for supporting dynamically sized chain listeners.
@@ -274,9 +251,8 @@ mod tests {
274251 ( chain. best_block_at_height( 2 ) , & listener_2 as & dyn chain:: Listen ) ,
275252 ( chain. best_block_at_height( 3 ) , & listener_3 as & dyn chain:: Listen ) ,
276253 ] ;
277- let mut cache = chain. header_cache ( 0 ..=4 ) ;
278- match synchronize_listeners ( & chain, Network :: Bitcoin , & mut cache, listeners) . await {
279- Ok ( header) => assert_eq ! ( header, chain. tip( ) ) ,
254+ match synchronize_listeners ( & chain, Network :: Bitcoin , listeners) . await {
255+ Ok ( ( _, header) ) => assert_eq ! ( header, chain. tip( ) ) ,
280256 Err ( e) => panic ! ( "Unexpected error: {:?}" , e) ,
281257 }
282258 }
@@ -306,11 +282,8 @@ mod tests {
306282 ( fork_chain_2. best_block( ) , & listener_2 as & dyn chain:: Listen ) ,
307283 ( fork_chain_3. best_block( ) , & listener_3 as & dyn chain:: Listen ) ,
308284 ] ;
309- let mut cache = fork_chain_1. header_cache ( 2 ..=4 ) ;
310- cache. extend ( fork_chain_2. header_cache ( 3 ..=4 ) ) ;
311- cache. extend ( fork_chain_3. header_cache ( 4 ..=4 ) ) ;
312- match synchronize_listeners ( & main_chain, Network :: Bitcoin , & mut cache, listeners) . await {
313- Ok ( header) => assert_eq ! ( header, main_chain. tip( ) ) ,
285+ match synchronize_listeners ( & main_chain, Network :: Bitcoin , listeners) . await {
286+ Ok ( ( _, header) ) => assert_eq ! ( header, main_chain. tip( ) ) ,
314287 Err ( e) => panic ! ( "Unexpected error: {:?}" , e) ,
315288 }
316289 }
@@ -343,33 +316,8 @@ mod tests {
343316 ( fork_chain_2. best_block( ) , & listener_2 as & dyn chain:: Listen ) ,
344317 ( fork_chain_3. best_block( ) , & listener_3 as & dyn chain:: Listen ) ,
345318 ] ;
346- let mut cache = fork_chain_1. header_cache ( 2 ..=4 ) ;
347- cache. extend ( fork_chain_2. header_cache ( 3 ..=4 ) ) ;
348- cache. extend ( fork_chain_3. header_cache ( 4 ..=4 ) ) ;
349- match synchronize_listeners ( & main_chain, Network :: Bitcoin , & mut cache, listeners) . await {
350- Ok ( header) => assert_eq ! ( header, main_chain. tip( ) ) ,
351- Err ( e) => panic ! ( "Unexpected error: {:?}" , e) ,
352- }
353- }
354-
355- #[ tokio:: test]
356- async fn cache_connected_and_keep_disconnected_blocks ( ) {
357- let main_chain = Blockchain :: default ( ) . with_height ( 2 ) ;
358- let fork_chain = main_chain. fork_at_height ( 1 ) ;
359- let new_tip = main_chain. tip ( ) ;
360- let old_best_block = fork_chain. best_block ( ) ;
361-
362- let listener = MockChainListener :: new ( )
363- . expect_blocks_disconnected ( * fork_chain. at_height ( 1 ) )
364- . expect_block_connected ( * new_tip) ;
365-
366- let listeners = vec ! [ ( old_best_block, & listener as & dyn chain:: Listen ) ] ;
367- let mut cache = fork_chain. header_cache ( 2 ..=2 ) ;
368- match synchronize_listeners ( & main_chain, Network :: Bitcoin , & mut cache, listeners) . await {
369- Ok ( _) => {
370- assert ! ( cache. contains_key( & new_tip. block_hash) ) ;
371- assert ! ( cache. contains_key( & old_best_block. block_hash) ) ;
372- } ,
319+ match synchronize_listeners ( & main_chain, Network :: Bitcoin , listeners) . await {
320+ Ok ( ( _, header) ) => assert_eq ! ( header, main_chain. tip( ) ) ,
373321 Err ( e) => panic ! ( "Unexpected error: {:?}" , e) ,
374322 }
375323 }
0 commit comments