]> git.bitcoin.ninja Git - rust-lightning/blob - lightning-background-processor/src/lib.rs
Fix bench build warning
[rust-lightning] / lightning-background-processor / src / lib.rs
1 //! Utilities that take care of tasks that (1) need to happen periodically to keep Rust-Lightning
2 //! running properly, and (2) either can or should be run in the background. See docs for
3 //! [`BackgroundProcessor`] for more details on the nitty-gritty.
4
5 #![deny(rustdoc::broken_intra_doc_links)]
6 #![deny(rustdoc::private_intra_doc_links)]
7
8 #![deny(missing_docs)]
9 #![cfg_attr(not(feature = "futures"), deny(unsafe_code))]
10
11 #![cfg_attr(docsrs, feature(doc_auto_cfg))]
12
13 #![cfg_attr(all(not(feature = "std"), not(test)), no_std)]
14
15 #[cfg(any(test, feature = "std"))]
16 extern crate core;
17
18 #[cfg(not(feature = "std"))]
19 extern crate alloc;
20
21 #[macro_use] extern crate lightning;
22 extern crate lightning_rapid_gossip_sync;
23
24 use lightning::chain;
25 use lightning::chain::chaininterface::{BroadcasterInterface, FeeEstimator};
26 use lightning::chain::chainmonitor::{ChainMonitor, Persist};
27 use lightning::sign::{EntropySource, NodeSigner, SignerProvider};
28 use lightning::events::{Event, PathFailure};
29 #[cfg(feature = "std")]
30 use lightning::events::EventHandler;
31 #[cfg(any(feature = "std", feature = "futures"))]
32 use lightning::events::EventsProvider;
33
34 use lightning::ln::channelmanager::ChannelManager;
35 use lightning::ln::msgs::OnionMessageHandler;
36 use lightning::ln::peer_handler::APeerManager;
37 use lightning::routing::gossip::{NetworkGraph, P2PGossipSync};
38 use lightning::routing::utxo::UtxoLookup;
39 use lightning::routing::router::Router;
40 use lightning::routing::scoring::{ScoreUpdate, WriteableScore};
41 use lightning::util::logger::Logger;
42 use lightning::util::persist::Persister;
43 #[cfg(feature = "std")]
44 use lightning::util::wakers::Sleeper;
45 use lightning_rapid_gossip_sync::RapidGossipSync;
46
47 use core::ops::Deref;
48 use core::time::Duration;
49
50 #[cfg(feature = "std")]
51 use std::sync::Arc;
52 #[cfg(feature = "std")]
53 use core::sync::atomic::{AtomicBool, Ordering};
54 #[cfg(feature = "std")]
55 use std::thread::{self, JoinHandle};
56 #[cfg(feature = "std")]
57 use std::time::Instant;
58
59 #[cfg(not(feature = "std"))]
60 use alloc::vec::Vec;
61
62 /// `BackgroundProcessor` takes care of tasks that (1) need to happen periodically to keep
63 /// Rust-Lightning running properly, and (2) either can or should be run in the background. Its
64 /// responsibilities are:
65 /// * Processing [`Event`]s with a user-provided [`EventHandler`].
66 /// * Monitoring whether the [`ChannelManager`] needs to be re-persisted to disk, and if so,
67 ///   writing it to disk/backups by invoking the callback given to it at startup.
68 ///   [`ChannelManager`] persistence should be done in the background.
69 /// * Calling [`ChannelManager::timer_tick_occurred`], [`ChainMonitor::rebroadcast_pending_claims`]
70 ///   and [`PeerManager::timer_tick_occurred`] at the appropriate intervals.
71 /// * Calling [`NetworkGraph::remove_stale_channels_and_tracking`] (if a [`GossipSync`] with a
72 ///   [`NetworkGraph`] is provided to [`BackgroundProcessor::start`]).
73 ///
74 /// It will also call [`PeerManager::process_events`] periodically though this shouldn't be relied
75 /// upon as doing so may result in high latency.
76 ///
77 /// # Note
78 ///
79 /// If [`ChannelManager`] persistence fails and the persisted manager becomes out-of-date, then
80 /// there is a risk of channels force-closing on startup when the manager realizes it's outdated.
81 /// However, as long as [`ChannelMonitor`] backups are sound, no funds besides those used for
82 /// unilateral chain closure fees are at risk.
83 ///
84 /// [`ChannelMonitor`]: lightning::chain::channelmonitor::ChannelMonitor
85 /// [`Event`]: lightning::events::Event
86 /// [`PeerManager::timer_tick_occurred`]: lightning::ln::peer_handler::PeerManager::timer_tick_occurred
87 /// [`PeerManager::process_events`]: lightning::ln::peer_handler::PeerManager::process_events
88 #[cfg(feature = "std")]
89 #[must_use = "BackgroundProcessor will immediately stop on drop. It should be stored until shutdown."]
90 pub struct BackgroundProcessor {
91         stop_thread: Arc<AtomicBool>,
92         thread_handle: Option<JoinHandle<Result<(), std::io::Error>>>,
93 }
94
95 #[cfg(not(test))]
96 const FRESHNESS_TIMER: u64 = 60;
97 #[cfg(test)]
98 const FRESHNESS_TIMER: u64 = 1;
99
100 #[cfg(all(not(test), not(debug_assertions)))]
101 const PING_TIMER: u64 = 10;
102 /// Signature operations take a lot longer without compiler optimisations.
103 /// Increasing the ping timer allows for this but slower devices will be disconnected if the
104 /// timeout is reached.
105 #[cfg(all(not(test), debug_assertions))]
106 const PING_TIMER: u64 = 30;
107 #[cfg(test)]
108 const PING_TIMER: u64 = 1;
109
110 #[cfg(not(test))]
111 const ONION_MESSAGE_HANDLER_TIMER: u64 = 10;
112 #[cfg(test)]
113 const ONION_MESSAGE_HANDLER_TIMER: u64 = 1;
114
115 /// Prune the network graph of stale entries hourly.
116 const NETWORK_PRUNE_TIMER: u64 = 60 * 60;
117
118 #[cfg(not(test))]
119 const SCORER_PERSIST_TIMER: u64 = 60 * 5;
120 #[cfg(test)]
121 const SCORER_PERSIST_TIMER: u64 = 1;
122
123 #[cfg(not(test))]
124 const FIRST_NETWORK_PRUNE_TIMER: u64 = 60;
125 #[cfg(test)]
126 const FIRST_NETWORK_PRUNE_TIMER: u64 = 1;
127
128 #[cfg(not(test))]
129 const REBROADCAST_TIMER: u64 = 30;
130 #[cfg(test)]
131 const REBROADCAST_TIMER: u64 = 1;
132
133 #[cfg(feature = "futures")]
134 /// core::cmp::min is not currently const, so we define a trivial (and equivalent) replacement
135 const fn min_u64(a: u64, b: u64) -> u64 { if a < b { a } else { b } }
136 #[cfg(feature = "futures")]
137 const FASTEST_TIMER: u64 = min_u64(min_u64(FRESHNESS_TIMER, PING_TIMER),
138         min_u64(SCORER_PERSIST_TIMER, min_u64(FIRST_NETWORK_PRUNE_TIMER, REBROADCAST_TIMER)));
139
140 /// Either [`P2PGossipSync`] or [`RapidGossipSync`].
141 pub enum GossipSync<
142         P: Deref<Target = P2PGossipSync<G, U, L>>,
143         R: Deref<Target = RapidGossipSync<G, L>>,
144         G: Deref<Target = NetworkGraph<L>>,
145         U: Deref,
146         L: Deref,
147 >
148 where U::Target: UtxoLookup, L::Target: Logger {
149         /// Gossip sync via the lightning peer-to-peer network as defined by BOLT 7.
150         P2P(P),
151         /// Rapid gossip sync from a trusted server.
152         Rapid(R),
153         /// No gossip sync.
154         None,
155 }
156
157 impl<
158         P: Deref<Target = P2PGossipSync<G, U, L>>,
159         R: Deref<Target = RapidGossipSync<G, L>>,
160         G: Deref<Target = NetworkGraph<L>>,
161         U: Deref,
162         L: Deref,
163 > GossipSync<P, R, G, U, L>
164 where U::Target: UtxoLookup, L::Target: Logger {
165         fn network_graph(&self) -> Option<&G> {
166                 match self {
167                         GossipSync::P2P(gossip_sync) => Some(gossip_sync.network_graph()),
168                         GossipSync::Rapid(gossip_sync) => Some(gossip_sync.network_graph()),
169                         GossipSync::None => None,
170                 }
171         }
172
173         fn prunable_network_graph(&self) -> Option<&G> {
174                 match self {
175                         GossipSync::P2P(gossip_sync) => Some(gossip_sync.network_graph()),
176                         GossipSync::Rapid(gossip_sync) => {
177                                 if gossip_sync.is_initial_sync_complete() {
178                                         Some(gossip_sync.network_graph())
179                                 } else {
180                                         None
181                                 }
182                         },
183                         GossipSync::None => None,
184                 }
185         }
186 }
187
188 /// This is not exported to bindings users as the bindings concretize everything and have constructors for us
189 impl<P: Deref<Target = P2PGossipSync<G, U, L>>, G: Deref<Target = NetworkGraph<L>>, U: Deref, L: Deref>
190         GossipSync<P, &RapidGossipSync<G, L>, G, U, L>
191 where
192         U::Target: UtxoLookup,
193         L::Target: Logger,
194 {
195         /// Initializes a new [`GossipSync::P2P`] variant.
196         pub fn p2p(gossip_sync: P) -> Self {
197                 GossipSync::P2P(gossip_sync)
198         }
199 }
200
201 /// This is not exported to bindings users as the bindings concretize everything and have constructors for us
202 impl<'a, R: Deref<Target = RapidGossipSync<G, L>>, G: Deref<Target = NetworkGraph<L>>, L: Deref>
203         GossipSync<
204                 &P2PGossipSync<G, &'a (dyn UtxoLookup + Send + Sync), L>,
205                 R,
206                 G,
207                 &'a (dyn UtxoLookup + Send + Sync),
208                 L,
209         >
210 where
211         L::Target: Logger,
212 {
213         /// Initializes a new [`GossipSync::Rapid`] variant.
214         pub fn rapid(gossip_sync: R) -> Self {
215                 GossipSync::Rapid(gossip_sync)
216         }
217 }
218
219 /// This is not exported to bindings users as the bindings concretize everything and have constructors for us
220 impl<'a, L: Deref>
221         GossipSync<
222                 &P2PGossipSync<&'a NetworkGraph<L>, &'a (dyn UtxoLookup + Send + Sync), L>,
223                 &RapidGossipSync<&'a NetworkGraph<L>, L>,
224                 &'a NetworkGraph<L>,
225                 &'a (dyn UtxoLookup + Send + Sync),
226                 L,
227         >
228 where
229         L::Target: Logger,
230 {
231         /// Initializes a new [`GossipSync::None`] variant.
232         pub fn none() -> Self {
233                 GossipSync::None
234         }
235 }
236
237 fn handle_network_graph_update<L: Deref>(
238         network_graph: &NetworkGraph<L>, event: &Event
239 ) where L::Target: Logger {
240         if let Event::PaymentPathFailed {
241                 failure: PathFailure::OnPath { network_update: Some(ref upd) }, .. } = event
242         {
243                 network_graph.handle_network_update(upd);
244         }
245 }
246
247 /// Updates scorer based on event and returns whether an update occurred so we can decide whether
248 /// to persist.
249 fn update_scorer<'a, S: 'static + Deref<Target = SC> + Send + Sync, SC: 'a + WriteableScore<'a>>(
250         scorer: &'a S, event: &Event, duration_since_epoch: Duration,
251 ) -> bool {
252         match event {
253                 Event::PaymentPathFailed { ref path, short_channel_id: Some(scid), .. } => {
254                         let mut score = scorer.write_lock();
255                         score.payment_path_failed(path, *scid, duration_since_epoch);
256                 },
257                 Event::PaymentPathFailed { ref path, payment_failed_permanently: true, .. } => {
258                         // Reached if the destination explicitly failed it back. We treat this as a successful probe
259                         // because the payment made it all the way to the destination with sufficient liquidity.
260                         let mut score = scorer.write_lock();
261                         score.probe_successful(path, duration_since_epoch);
262                 },
263                 Event::PaymentPathSuccessful { path, .. } => {
264                         let mut score = scorer.write_lock();
265                         score.payment_path_successful(path, duration_since_epoch);
266                 },
267                 Event::ProbeSuccessful { path, .. } => {
268                         let mut score = scorer.write_lock();
269                         score.probe_successful(path, duration_since_epoch);
270                 },
271                 Event::ProbeFailed { path, short_channel_id: Some(scid), .. } => {
272                         let mut score = scorer.write_lock();
273                         score.probe_failed(path, *scid, duration_since_epoch);
274                 },
275                 _ => return false,
276         }
277         true
278 }
279
280 macro_rules! define_run_body {
281         (
282                 $persister: ident, $chain_monitor: ident, $process_chain_monitor_events: expr,
283                 $channel_manager: ident, $process_channel_manager_events: expr,
284                 $peer_manager: ident, $process_onion_message_handler_events: expr, $gossip_sync: ident,
285                 $logger: ident, $scorer: ident, $loop_exit_check: expr, $await: expr, $get_timer: expr,
286                 $timer_elapsed: expr, $check_slow_await: expr, $time_fetch: expr,
287         ) => { {
288                 log_trace!($logger, "Calling ChannelManager's timer_tick_occurred on startup");
289                 $channel_manager.timer_tick_occurred();
290                 log_trace!($logger, "Rebroadcasting monitor's pending claims on startup");
291                 $chain_monitor.rebroadcast_pending_claims();
292
293                 let mut last_freshness_call = $get_timer(FRESHNESS_TIMER);
294                 let mut last_onion_message_handler_call = $get_timer(ONION_MESSAGE_HANDLER_TIMER);
295                 let mut last_ping_call = $get_timer(PING_TIMER);
296                 let mut last_prune_call = $get_timer(FIRST_NETWORK_PRUNE_TIMER);
297                 let mut last_scorer_persist_call = $get_timer(SCORER_PERSIST_TIMER);
298                 let mut last_rebroadcast_call = $get_timer(REBROADCAST_TIMER);
299                 let mut have_pruned = false;
300                 let mut have_decayed_scorer = false;
301
302                 loop {
303                         $process_channel_manager_events;
304                         $process_chain_monitor_events;
305                         $process_onion_message_handler_events;
306
307                         // Note that the PeerManager::process_events may block on ChannelManager's locks,
308                         // hence it comes last here. When the ChannelManager finishes whatever it's doing,
309                         // we want to ensure we get into `persist_manager` as quickly as we can, especially
310                         // without running the normal event processing above and handing events to users.
311                         //
312                         // Specifically, on an *extremely* slow machine, we may see ChannelManager start
313                         // processing a message effectively at any point during this loop. In order to
314                         // minimize the time between such processing completing and persisting the updated
315                         // ChannelManager, we want to minimize methods blocking on a ChannelManager
316                         // generally, and as a fallback place such blocking only immediately before
317                         // persistence.
318                         $peer_manager.as_ref().process_events();
319
320                         // Exit the loop if the background processor was requested to stop.
321                         if $loop_exit_check {
322                                 log_trace!($logger, "Terminating background processor.");
323                                 break;
324                         }
325
326                         // We wait up to 100ms, but track how long it takes to detect being put to sleep,
327                         // see `await_start`'s use below.
328                         let mut await_start = None;
329                         if $check_slow_await { await_start = Some($get_timer(1)); }
330                         $await;
331                         let await_slow = if $check_slow_await { $timer_elapsed(&mut await_start.unwrap(), 1) } else { false };
332
333                         // Exit the loop if the background processor was requested to stop.
334                         if $loop_exit_check {
335                                 log_trace!($logger, "Terminating background processor.");
336                                 break;
337                         }
338
339                         if $channel_manager.get_and_clear_needs_persistence() {
340                                 log_trace!($logger, "Persisting ChannelManager...");
341                                 $persister.persist_manager(&*$channel_manager)?;
342                                 log_trace!($logger, "Done persisting ChannelManager.");
343                         }
344                         if $timer_elapsed(&mut last_freshness_call, FRESHNESS_TIMER) {
345                                 log_trace!($logger, "Calling ChannelManager's timer_tick_occurred");
346                                 $channel_manager.timer_tick_occurred();
347                                 last_freshness_call = $get_timer(FRESHNESS_TIMER);
348                         }
349                         if $timer_elapsed(&mut last_onion_message_handler_call, ONION_MESSAGE_HANDLER_TIMER) {
350                                 log_trace!($logger, "Calling OnionMessageHandler's timer_tick_occurred");
351                                 $peer_manager.onion_message_handler().timer_tick_occurred();
352                                 last_onion_message_handler_call = $get_timer(ONION_MESSAGE_HANDLER_TIMER);
353                         }
354                         if await_slow {
355                                 // On various platforms, we may be starved of CPU cycles for several reasons.
356                                 // E.g. on iOS, if we've been in the background, we will be entirely paused.
357                                 // Similarly, if we're on a desktop platform and the device has been asleep, we
358                                 // may not get any cycles.
359                                 // We detect this by checking if our max-100ms-sleep, above, ran longer than a
360                                 // full second, at which point we assume sockets may have been killed (they
361                                 // appear to be at least on some platforms, even if it has only been a second).
362                                 // Note that we have to take care to not get here just because user event
363                                 // processing was slow at the top of the loop. For example, the sample client
364                                 // may call Bitcoin Core RPCs during event handling, which very often takes
365                                 // more than a handful of seconds to complete, and shouldn't disconnect all our
366                                 // peers.
367                                 log_trace!($logger, "100ms sleep took more than a second, disconnecting peers.");
368                                 $peer_manager.as_ref().disconnect_all_peers();
369                                 last_ping_call = $get_timer(PING_TIMER);
370                         } else if $timer_elapsed(&mut last_ping_call, PING_TIMER) {
371                                 log_trace!($logger, "Calling PeerManager's timer_tick_occurred");
372                                 $peer_manager.as_ref().timer_tick_occurred();
373                                 last_ping_call = $get_timer(PING_TIMER);
374                         }
375
376                         // Note that we want to run a graph prune once not long after startup before
377                         // falling back to our usual hourly prunes. This avoids short-lived clients never
378                         // pruning their network graph. We run once 60 seconds after startup before
379                         // continuing our normal cadence. For RGS, since 60 seconds is likely too long,
380                         // we prune after an initial sync completes.
381                         let prune_timer = if have_pruned { NETWORK_PRUNE_TIMER } else { FIRST_NETWORK_PRUNE_TIMER };
382                         let prune_timer_elapsed = $timer_elapsed(&mut last_prune_call, prune_timer);
383                         let should_prune = match $gossip_sync {
384                                 GossipSync::Rapid(_) => !have_pruned || prune_timer_elapsed,
385                                 _ => prune_timer_elapsed,
386                         };
387                         if should_prune {
388                                 // The network graph must not be pruned while rapid sync completion is pending
389                                 if let Some(network_graph) = $gossip_sync.prunable_network_graph() {
390                                         if let Some(duration_since_epoch) = $time_fetch() {
391                                                 log_trace!($logger, "Pruning and persisting network graph.");
392                                                 network_graph.remove_stale_channels_and_tracking_with_time(duration_since_epoch.as_secs());
393                                         } else {
394                                                 log_warn!($logger, "Not pruning network graph, consider enabling `std` or doing so manually with remove_stale_channels_and_tracking_with_time.");
395                                                 log_trace!($logger, "Persisting network graph.");
396                                         }
397
398                                         if let Err(e) = $persister.persist_graph(network_graph) {
399                                                 log_error!($logger, "Error: Failed to persist network graph, check your disk and permissions {}", e)
400                                         }
401
402                                         have_pruned = true;
403                                 }
404                                 let prune_timer = if have_pruned { NETWORK_PRUNE_TIMER } else { FIRST_NETWORK_PRUNE_TIMER };
405                                 last_prune_call = $get_timer(prune_timer);
406                         }
407
408                         if !have_decayed_scorer {
409                                 if let Some(ref scorer) = $scorer {
410                                         if let Some(duration_since_epoch) = $time_fetch() {
411                                                 log_trace!($logger, "Calling time_passed on scorer at startup");
412                                                 scorer.write_lock().time_passed(duration_since_epoch);
413                                         }
414                                 }
415                                 have_decayed_scorer = true;
416                         }
417
418                         if $timer_elapsed(&mut last_scorer_persist_call, SCORER_PERSIST_TIMER) {
419                                 if let Some(ref scorer) = $scorer {
420                                         if let Some(duration_since_epoch) = $time_fetch() {
421                                                 log_trace!($logger, "Calling time_passed and persisting scorer");
422                                                 scorer.write_lock().time_passed(duration_since_epoch);
423                                         } else {
424                                                 log_trace!($logger, "Persisting scorer");
425                                         }
426                                         if let Err(e) = $persister.persist_scorer(&scorer) {
427                                                 log_error!($logger, "Error: Failed to persist scorer, check your disk and permissions {}", e)
428                                         }
429                                 }
430                                 last_scorer_persist_call = $get_timer(SCORER_PERSIST_TIMER);
431                         }
432
433                         if $timer_elapsed(&mut last_rebroadcast_call, REBROADCAST_TIMER) {
434                                 log_trace!($logger, "Rebroadcasting monitor's pending claims");
435                                 $chain_monitor.rebroadcast_pending_claims();
436                                 last_rebroadcast_call = $get_timer(REBROADCAST_TIMER);
437                         }
438                 }
439
440                 // After we exit, ensure we persist the ChannelManager one final time - this avoids
441                 // some races where users quit while channel updates were in-flight, with
442                 // ChannelMonitor update(s) persisted without a corresponding ChannelManager update.
443                 $persister.persist_manager(&*$channel_manager)?;
444
445                 // Persist Scorer on exit
446                 if let Some(ref scorer) = $scorer {
447                         $persister.persist_scorer(&scorer)?;
448                 }
449
450                 // Persist NetworkGraph on exit
451                 if let Some(network_graph) = $gossip_sync.network_graph() {
452                         $persister.persist_graph(network_graph)?;
453                 }
454
455                 Ok(())
456         } }
457 }
458
459 #[cfg(feature = "futures")]
460 pub(crate) mod futures_util {
461         use core::future::Future;
462         use core::task::{Poll, Waker, RawWaker, RawWakerVTable};
463         use core::pin::Pin;
464         use core::marker::Unpin;
465         pub(crate) struct Selector<
466                 A: Future<Output=()> + Unpin, B: Future<Output=()> + Unpin, C: Future<Output=bool> + Unpin
467         > {
468                 pub a: A,
469                 pub b: B,
470                 pub c: C,
471         }
472         pub(crate) enum SelectorOutput {
473                 A, B, C(bool),
474         }
475
476         impl<
477                 A: Future<Output=()> + Unpin, B: Future<Output=()> + Unpin, C: Future<Output=bool> + Unpin
478         > Future for Selector<A, B, C> {
479                 type Output = SelectorOutput;
480                 fn poll(mut self: Pin<&mut Self>, ctx: &mut core::task::Context<'_>) -> Poll<SelectorOutput> {
481                         match Pin::new(&mut self.a).poll(ctx) {
482                                 Poll::Ready(()) => { return Poll::Ready(SelectorOutput::A); },
483                                 Poll::Pending => {},
484                         }
485                         match Pin::new(&mut self.b).poll(ctx) {
486                                 Poll::Ready(()) => { return Poll::Ready(SelectorOutput::B); },
487                                 Poll::Pending => {},
488                         }
489                         match Pin::new(&mut self.c).poll(ctx) {
490                                 Poll::Ready(res) => { return Poll::Ready(SelectorOutput::C(res)); },
491                                 Poll::Pending => {},
492                         }
493                         Poll::Pending
494                 }
495         }
496
497         // If we want to poll a future without an async context to figure out if it has completed or
498         // not without awaiting, we need a Waker, which needs a vtable...we fill it with dummy values
499         // but sadly there's a good bit of boilerplate here.
500         fn dummy_waker_clone(_: *const ()) -> RawWaker { RawWaker::new(core::ptr::null(), &DUMMY_WAKER_VTABLE) }
501         fn dummy_waker_action(_: *const ()) { }
502
503         const DUMMY_WAKER_VTABLE: RawWakerVTable = RawWakerVTable::new(
504                 dummy_waker_clone, dummy_waker_action, dummy_waker_action, dummy_waker_action);
505         pub(crate) fn dummy_waker() -> Waker { unsafe { Waker::from_raw(RawWaker::new(core::ptr::null(), &DUMMY_WAKER_VTABLE)) } }
506 }
507 #[cfg(feature = "futures")]
508 use futures_util::{Selector, SelectorOutput, dummy_waker};
509 #[cfg(feature = "futures")]
510 use core::task;
511
512 /// Processes background events in a future.
513 ///
514 /// `sleeper` should return a future which completes in the given amount of time and returns a
515 /// boolean indicating whether the background processing should exit. Once `sleeper` returns a
516 /// future which outputs `true`, the loop will exit and this function's future will complete.
517 /// The `sleeper` future is free to return early after it has triggered the exit condition.
518 ///
519 /// See [`BackgroundProcessor::start`] for information on which actions this handles.
520 ///
521 /// Requires the `futures` feature. Note that while this method is available without the `std`
522 /// feature, doing so will skip calling [`NetworkGraph::remove_stale_channels_and_tracking`],
523 /// you should call [`NetworkGraph::remove_stale_channels_and_tracking_with_time`] regularly
524 /// manually instead.
525 ///
526 /// The `mobile_interruptable_platform` flag should be set if we're currently running on a
527 /// mobile device, where we may need to check for interruption of the application regularly. If you
528 /// are unsure, you should set the flag, as the performance impact of it is minimal unless there
529 /// are hundreds or thousands of simultaneous process calls running.
530 ///
531 /// The `fetch_time` parameter should return the current wall clock time, if one is available. If
532 /// no time is available, some features may be disabled, however the node will still operate fine.
533 ///
534 /// For example, in order to process background events in a [Tokio](https://tokio.rs/) task, you
535 /// could setup `process_events_async` like this:
536 /// ```
537 /// # use lightning::io;
538 /// # use std::sync::{Arc, RwLock};
539 /// # use std::sync::atomic::{AtomicBool, Ordering};
540 /// # use std::time::SystemTime;
541 /// # use lightning_background_processor::{process_events_async, GossipSync};
542 /// # struct MyStore {}
543 /// # impl lightning::util::persist::KVStore for MyStore {
544 /// #     fn read(&self, primary_namespace: &str, secondary_namespace: &str, key: &str) -> io::Result<Vec<u8>> { Ok(Vec::new()) }
545 /// #     fn write(&self, primary_namespace: &str, secondary_namespace: &str, key: &str, buf: &[u8]) -> io::Result<()> { Ok(()) }
546 /// #     fn remove(&self, primary_namespace: &str, secondary_namespace: &str, key: &str, lazy: bool) -> io::Result<()> { Ok(()) }
547 /// #     fn list(&self, primary_namespace: &str, secondary_namespace: &str) -> io::Result<Vec<String>> { Ok(Vec::new()) }
548 /// # }
549 /// # struct MyEventHandler {}
550 /// # impl MyEventHandler {
551 /// #     async fn handle_event(&self, _: lightning::events::Event) {}
552 /// # }
553 /// # #[derive(Eq, PartialEq, Clone, Hash)]
554 /// # struct MySocketDescriptor {}
555 /// # impl lightning::ln::peer_handler::SocketDescriptor for MySocketDescriptor {
556 /// #     fn send_data(&mut self, _data: &[u8], _resume_read: bool) -> usize { 0 }
557 /// #     fn disconnect_socket(&mut self) {}
558 /// # }
559 /// # type MyBroadcaster = dyn lightning::chain::chaininterface::BroadcasterInterface + Send + Sync;
560 /// # type MyFeeEstimator = dyn lightning::chain::chaininterface::FeeEstimator + Send + Sync;
561 /// # type MyNodeSigner = dyn lightning::sign::NodeSigner + Send + Sync;
562 /// # type MyUtxoLookup = dyn lightning::routing::utxo::UtxoLookup + Send + Sync;
563 /// # type MyFilter = dyn lightning::chain::Filter + Send + Sync;
564 /// # type MyLogger = dyn lightning::util::logger::Logger + Send + Sync;
565 /// # type MyChainMonitor = lightning::chain::chainmonitor::ChainMonitor<lightning::sign::InMemorySigner, Arc<MyFilter>, Arc<MyBroadcaster>, Arc<MyFeeEstimator>, Arc<MyLogger>, Arc<MyStore>>;
566 /// # type MyPeerManager = lightning::ln::peer_handler::SimpleArcPeerManager<MySocketDescriptor, MyChainMonitor, MyBroadcaster, MyFeeEstimator, Arc<MyUtxoLookup>, MyLogger>;
567 /// # type MyNetworkGraph = lightning::routing::gossip::NetworkGraph<Arc<MyLogger>>;
568 /// # type MyGossipSync = lightning::routing::gossip::P2PGossipSync<Arc<MyNetworkGraph>, Arc<MyUtxoLookup>, Arc<MyLogger>>;
569 /// # type MyChannelManager = lightning::ln::channelmanager::SimpleArcChannelManager<MyChainMonitor, MyBroadcaster, MyFeeEstimator, MyLogger>;
570 /// # type MyScorer = RwLock<lightning::routing::scoring::ProbabilisticScorer<Arc<MyNetworkGraph>, Arc<MyLogger>>>;
571 ///
572 /// # async fn setup_background_processing(my_persister: Arc<MyStore>, my_event_handler: Arc<MyEventHandler>, my_chain_monitor: Arc<MyChainMonitor>, my_channel_manager: Arc<MyChannelManager>, my_gossip_sync: Arc<MyGossipSync>, my_logger: Arc<MyLogger>, my_scorer: Arc<MyScorer>, my_peer_manager: Arc<MyPeerManager>) {
573 ///     let background_persister = Arc::clone(&my_persister);
574 ///     let background_event_handler = Arc::clone(&my_event_handler);
575 ///     let background_chain_mon = Arc::clone(&my_chain_monitor);
576 ///     let background_chan_man = Arc::clone(&my_channel_manager);
577 ///     let background_gossip_sync = GossipSync::p2p(Arc::clone(&my_gossip_sync));
578 ///     let background_peer_man = Arc::clone(&my_peer_manager);
579 ///     let background_logger = Arc::clone(&my_logger);
580 ///     let background_scorer = Arc::clone(&my_scorer);
581 ///
582 ///     // Setup the sleeper.
583 ///     let (stop_sender, stop_receiver) = tokio::sync::watch::channel(());
584 ///
585 ///     let sleeper = move |d| {
586 ///             let mut receiver = stop_receiver.clone();
587 ///             Box::pin(async move {
588 ///                     tokio::select!{
589 ///                             _ = tokio::time::sleep(d) => false,
590 ///                             _ = receiver.changed() => true,
591 ///                     }
592 ///             })
593 ///     };
594 ///
595 ///     let mobile_interruptable_platform = false;
596 ///
597 ///     let handle = tokio::spawn(async move {
598 ///             process_events_async(
599 ///                     background_persister,
600 ///                     |e| background_event_handler.handle_event(e),
601 ///                     background_chain_mon,
602 ///                     background_chan_man,
603 ///                     background_gossip_sync,
604 ///                     background_peer_man,
605 ///                     background_logger,
606 ///                     Some(background_scorer),
607 ///                     sleeper,
608 ///                     mobile_interruptable_platform,
609 ///                     || Some(SystemTime::now().duration_since(SystemTime::UNIX_EPOCH).unwrap())
610 ///                     )
611 ///                     .await
612 ///                     .expect("Failed to process events");
613 ///     });
614 ///
615 ///     // Stop the background processing.
616 ///     stop_sender.send(()).unwrap();
617 ///     handle.await.unwrap();
618 ///     # }
619 ///```
620 #[cfg(feature = "futures")]
621 pub async fn process_events_async<
622         'a,
623         UL: 'static + Deref + Send + Sync,
624         CF: 'static + Deref + Send + Sync,
625         CW: 'static + Deref + Send + Sync,
626         T: 'static + Deref + Send + Sync,
627         ES: 'static + Deref + Send + Sync,
628         NS: 'static + Deref + Send + Sync,
629         SP: 'static + Deref + Send + Sync,
630         F: 'static + Deref + Send + Sync,
631         R: 'static + Deref + Send + Sync,
632         G: 'static + Deref<Target = NetworkGraph<L>> + Send + Sync,
633         L: 'static + Deref + Send + Sync,
634         P: 'static + Deref + Send + Sync,
635         EventHandlerFuture: core::future::Future<Output = ()>,
636         EventHandler: Fn(Event) -> EventHandlerFuture,
637         PS: 'static + Deref + Send,
638         M: 'static + Deref<Target = ChainMonitor<<SP::Target as SignerProvider>::EcdsaSigner, CF, T, F, L, P>> + Send + Sync,
639         CM: 'static + Deref<Target = ChannelManager<CW, T, ES, NS, SP, F, R, L>> + Send + Sync,
640         PGS: 'static + Deref<Target = P2PGossipSync<G, UL, L>> + Send + Sync,
641         RGS: 'static + Deref<Target = RapidGossipSync<G, L>> + Send,
642         PM: 'static + Deref + Send + Sync,
643         S: 'static + Deref<Target = SC> + Send + Sync,
644         SC: for<'b> WriteableScore<'b>,
645         SleepFuture: core::future::Future<Output = bool> + core::marker::Unpin,
646         Sleeper: Fn(Duration) -> SleepFuture,
647         FetchTime: Fn() -> Option<Duration>,
648 >(
649         persister: PS, event_handler: EventHandler, chain_monitor: M, channel_manager: CM,
650         gossip_sync: GossipSync<PGS, RGS, G, UL, L>, peer_manager: PM, logger: L, scorer: Option<S>,
651         sleeper: Sleeper, mobile_interruptable_platform: bool, fetch_time: FetchTime,
652 ) -> Result<(), lightning::io::Error>
653 where
654         UL::Target: 'static + UtxoLookup,
655         CF::Target: 'static + chain::Filter,
656         CW::Target: 'static + chain::Watch<<SP::Target as SignerProvider>::EcdsaSigner>,
657         T::Target: 'static + BroadcasterInterface,
658         ES::Target: 'static + EntropySource,
659         NS::Target: 'static + NodeSigner,
660         SP::Target: 'static + SignerProvider,
661         F::Target: 'static + FeeEstimator,
662         R::Target: 'static + Router,
663         L::Target: 'static + Logger,
664         P::Target: 'static + Persist<<SP::Target as SignerProvider>::EcdsaSigner>,
665         PS::Target: 'static + Persister<'a, CW, T, ES, NS, SP, F, R, L, SC>,
666         PM::Target: APeerManager + Send + Sync,
667 {
668         let mut should_break = false;
669         let async_event_handler = |event| {
670                 let network_graph = gossip_sync.network_graph();
671                 let event_handler = &event_handler;
672                 let scorer = &scorer;
673                 let logger = &logger;
674                 let persister = &persister;
675                 let fetch_time = &fetch_time;
676                 async move {
677                         if let Some(network_graph) = network_graph {
678                                 handle_network_graph_update(network_graph, &event)
679                         }
680                         if let Some(ref scorer) = scorer {
681                                 if let Some(duration_since_epoch) = fetch_time() {
682                                         if update_scorer(scorer, &event, duration_since_epoch) {
683                                                 log_trace!(logger, "Persisting scorer after update");
684                                                 if let Err(e) = persister.persist_scorer(&scorer) {
685                                                         log_error!(logger, "Error: Failed to persist scorer, check your disk and permissions {}", e)
686                                                 }
687                                         }
688                                 }
689                         }
690                         event_handler(event).await;
691                 }
692         };
693         define_run_body!(
694                 persister, chain_monitor,
695                 chain_monitor.process_pending_events_async(async_event_handler).await,
696                 channel_manager, channel_manager.process_pending_events_async(async_event_handler).await,
697                 peer_manager, process_onion_message_handler_events_async(&peer_manager, async_event_handler).await,
698                 gossip_sync, logger, scorer, should_break, {
699                         let fut = Selector {
700                                 a: channel_manager.get_event_or_persistence_needed_future(),
701                                 b: chain_monitor.get_update_future(),
702                                 c: sleeper(if mobile_interruptable_platform { Duration::from_millis(100) } else { Duration::from_secs(FASTEST_TIMER) }),
703                         };
704                         match fut.await {
705                                 SelectorOutput::A|SelectorOutput::B => {},
706                                 SelectorOutput::C(exit) => {
707                                         should_break = exit;
708                                 }
709                         }
710                 }, |t| sleeper(Duration::from_secs(t)),
711                 |fut: &mut SleepFuture, _| {
712                         let mut waker = dummy_waker();
713                         let mut ctx = task::Context::from_waker(&mut waker);
714                         match core::pin::Pin::new(fut).poll(&mut ctx) {
715                                 task::Poll::Ready(exit) => { should_break = exit; true },
716                                 task::Poll::Pending => false,
717                         }
718                 }, mobile_interruptable_platform, fetch_time,
719         )
720 }
721
722 #[cfg(feature = "futures")]
723 async fn process_onion_message_handler_events_async<
724         EventHandlerFuture: core::future::Future<Output = ()>,
725         EventHandler: Fn(Event) -> EventHandlerFuture,
726         PM: 'static + Deref + Send + Sync,
727 >(
728         peer_manager: &PM, handler: EventHandler
729 )
730 where
731         PM::Target: APeerManager + Send + Sync,
732 {
733         let events = core::cell::RefCell::new(Vec::new());
734         peer_manager.onion_message_handler().process_pending_events(&|e| events.borrow_mut().push(e));
735
736         for event in events.into_inner() {
737                 handler(event).await
738         }
739 }
740
741 #[cfg(feature = "std")]
742 impl BackgroundProcessor {
743         /// Start a background thread that takes care of responsibilities enumerated in the [top-level
744         /// documentation].
745         ///
746         /// The thread runs indefinitely unless the object is dropped, [`stop`] is called, or
747         /// [`Persister::persist_manager`] returns an error. In case of an error, the error is retrieved by calling
748         /// either [`join`] or [`stop`].
749         ///
750         /// # Data Persistence
751         ///
752         /// [`Persister::persist_manager`] is responsible for writing out the [`ChannelManager`] to disk, and/or
753         /// uploading to one or more backup services. See [`ChannelManager::write`] for writing out a
754         /// [`ChannelManager`]. See the `lightning-persister` crate for LDK's
755         /// provided implementation.
756         ///
757         /// [`Persister::persist_graph`] is responsible for writing out the [`NetworkGraph`] to disk, if
758         /// [`GossipSync`] is supplied. See [`NetworkGraph::write`] for writing out a [`NetworkGraph`].
759         /// See the `lightning-persister` crate for LDK's provided implementation.
760         ///
761         /// Typically, users should either implement [`Persister::persist_manager`] to never return an
762         /// error or call [`join`] and handle any error that may arise. For the latter case,
763         /// `BackgroundProcessor` must be restarted by calling `start` again after handling the error.
764         ///
765         /// # Event Handling
766         ///
767         /// `event_handler` is responsible for handling events that users should be notified of (e.g.,
768         /// payment failed). [`BackgroundProcessor`] may decorate the given [`EventHandler`] with common
769         /// functionality implemented by other handlers.
770         /// * [`P2PGossipSync`] if given will update the [`NetworkGraph`] based on payment failures.
771         ///
772         /// # Rapid Gossip Sync
773         ///
774         /// If rapid gossip sync is meant to run at startup, pass [`RapidGossipSync`] via `gossip_sync`
775         /// to indicate that the [`BackgroundProcessor`] should not prune the [`NetworkGraph`] instance
776         /// until the [`RapidGossipSync`] instance completes its first sync.
777         ///
778         /// [top-level documentation]: BackgroundProcessor
779         /// [`join`]: Self::join
780         /// [`stop`]: Self::stop
781         /// [`ChannelManager`]: lightning::ln::channelmanager::ChannelManager
782         /// [`ChannelManager::write`]: lightning::ln::channelmanager::ChannelManager#impl-Writeable
783         /// [`Persister::persist_manager`]: lightning::util::persist::Persister::persist_manager
784         /// [`Persister::persist_graph`]: lightning::util::persist::Persister::persist_graph
785         /// [`NetworkGraph`]: lightning::routing::gossip::NetworkGraph
786         /// [`NetworkGraph::write`]: lightning::routing::gossip::NetworkGraph#impl-Writeable
787         pub fn start<
788                 'a,
789                 UL: 'static + Deref + Send + Sync,
790                 CF: 'static + Deref + Send + Sync,
791                 CW: 'static + Deref + Send + Sync,
792                 T: 'static + Deref + Send + Sync,
793                 ES: 'static + Deref + Send + Sync,
794                 NS: 'static + Deref + Send + Sync,
795                 SP: 'static + Deref + Send + Sync,
796                 F: 'static + Deref + Send + Sync,
797                 R: 'static + Deref + Send + Sync,
798                 G: 'static + Deref<Target = NetworkGraph<L>> + Send + Sync,
799                 L: 'static + Deref + Send + Sync,
800                 P: 'static + Deref + Send + Sync,
801                 EH: 'static + EventHandler + Send,
802                 PS: 'static + Deref + Send,
803                 M: 'static + Deref<Target = ChainMonitor<<SP::Target as SignerProvider>::EcdsaSigner, CF, T, F, L, P>> + Send + Sync,
804                 CM: 'static + Deref<Target = ChannelManager<CW, T, ES, NS, SP, F, R, L>> + Send + Sync,
805                 PGS: 'static + Deref<Target = P2PGossipSync<G, UL, L>> + Send + Sync,
806                 RGS: 'static + Deref<Target = RapidGossipSync<G, L>> + Send,
807                 PM: 'static + Deref + Send + Sync,
808                 S: 'static + Deref<Target = SC> + Send + Sync,
809                 SC: for <'b> WriteableScore<'b>,
810         >(
811                 persister: PS, event_handler: EH, chain_monitor: M, channel_manager: CM,
812                 gossip_sync: GossipSync<PGS, RGS, G, UL, L>, peer_manager: PM, logger: L, scorer: Option<S>,
813         ) -> Self
814         where
815                 UL::Target: 'static + UtxoLookup,
816                 CF::Target: 'static + chain::Filter,
817                 CW::Target: 'static + chain::Watch<<SP::Target as SignerProvider>::EcdsaSigner>,
818                 T::Target: 'static + BroadcasterInterface,
819                 ES::Target: 'static + EntropySource,
820                 NS::Target: 'static + NodeSigner,
821                 SP::Target: 'static + SignerProvider,
822                 F::Target: 'static + FeeEstimator,
823                 R::Target: 'static + Router,
824                 L::Target: 'static + Logger,
825                 P::Target: 'static + Persist<<SP::Target as SignerProvider>::EcdsaSigner>,
826                 PS::Target: 'static + Persister<'a, CW, T, ES, NS, SP, F, R, L, SC>,
827                 PM::Target: APeerManager + Send + Sync,
828         {
829                 let stop_thread = Arc::new(AtomicBool::new(false));
830                 let stop_thread_clone = stop_thread.clone();
831                 let handle = thread::spawn(move || -> Result<(), std::io::Error> {
832                         let event_handler = |event| {
833                                 let network_graph = gossip_sync.network_graph();
834                                 if let Some(network_graph) = network_graph {
835                                         handle_network_graph_update(network_graph, &event)
836                                 }
837                                 if let Some(ref scorer) = scorer {
838                                         use std::time::SystemTime;
839                                         let duration_since_epoch = SystemTime::now().duration_since(SystemTime::UNIX_EPOCH)
840                                                 .expect("Time should be sometime after 1970");
841                                         if update_scorer(scorer, &event, duration_since_epoch) {
842                                                 log_trace!(logger, "Persisting scorer after update");
843                                                 if let Err(e) = persister.persist_scorer(&scorer) {
844                                                         log_error!(logger, "Error: Failed to persist scorer, check your disk and permissions {}", e)
845                                                 }
846                                         }
847                                 }
848                                 event_handler.handle_event(event);
849                         };
850                         define_run_body!(
851                                 persister, chain_monitor, chain_monitor.process_pending_events(&event_handler),
852                                 channel_manager, channel_manager.process_pending_events(&event_handler),
853                                 peer_manager,
854                                 peer_manager.onion_message_handler().process_pending_events(&event_handler),
855                                 gossip_sync, logger, scorer, stop_thread.load(Ordering::Acquire),
856                                 { Sleeper::from_two_futures(
857                                         channel_manager.get_event_or_persistence_needed_future(),
858                                         chain_monitor.get_update_future()
859                                 ).wait_timeout(Duration::from_millis(100)); },
860                                 |_| Instant::now(), |time: &Instant, dur| time.elapsed().as_secs() > dur, false,
861                                 || {
862                                         use std::time::SystemTime;
863                                         Some(SystemTime::now().duration_since(SystemTime::UNIX_EPOCH)
864                                                 .expect("Time should be sometime after 1970"))
865                                 },
866                         )
867                 });
868                 Self { stop_thread: stop_thread_clone, thread_handle: Some(handle) }
869         }
870
871         /// Join `BackgroundProcessor`'s thread, returning any error that occurred while persisting
872         /// [`ChannelManager`].
873         ///
874         /// # Panics
875         ///
876         /// This function panics if the background thread has panicked such as while persisting or
877         /// handling events.
878         ///
879         /// [`ChannelManager`]: lightning::ln::channelmanager::ChannelManager
880         pub fn join(mut self) -> Result<(), std::io::Error> {
881                 assert!(self.thread_handle.is_some());
882                 self.join_thread()
883         }
884
885         /// Stop `BackgroundProcessor`'s thread, returning any error that occurred while persisting
886         /// [`ChannelManager`].
887         ///
888         /// # Panics
889         ///
890         /// This function panics if the background thread has panicked such as while persisting or
891         /// handling events.
892         ///
893         /// [`ChannelManager`]: lightning::ln::channelmanager::ChannelManager
894         pub fn stop(mut self) -> Result<(), std::io::Error> {
895                 assert!(self.thread_handle.is_some());
896                 self.stop_and_join_thread()
897         }
898
899         fn stop_and_join_thread(&mut self) -> Result<(), std::io::Error> {
900                 self.stop_thread.store(true, Ordering::Release);
901                 self.join_thread()
902         }
903
904         fn join_thread(&mut self) -> Result<(), std::io::Error> {
905                 match self.thread_handle.take() {
906                         Some(handle) => handle.join().unwrap(),
907                         None => Ok(()),
908                 }
909         }
910 }
911
912 #[cfg(feature = "std")]
913 impl Drop for BackgroundProcessor {
914         fn drop(&mut self) {
915                 self.stop_and_join_thread().unwrap();
916         }
917 }
918
919 #[cfg(all(feature = "std", test))]
920 mod tests {
921         use bitcoin::blockdata::constants::{genesis_block, ChainHash};
922         use bitcoin::blockdata::locktime::absolute::LockTime;
923         use bitcoin::blockdata::transaction::{Transaction, TxOut};
924         use bitcoin::network::constants::Network;
925         use bitcoin::secp256k1::{SecretKey, PublicKey, Secp256k1};
926         use lightning::chain::{BestBlock, Confirm, chainmonitor};
927         use lightning::chain::channelmonitor::ANTI_REORG_DELAY;
928         use lightning::sign::{InMemorySigner, KeysManager};
929         use lightning::chain::transaction::OutPoint;
930         use lightning::events::{Event, PathFailure, MessageSendEventsProvider, MessageSendEvent};
931         use lightning::{get_event_msg, get_event};
932         use lightning::ln::PaymentHash;
933         use lightning::ln::channelmanager;
934         use lightning::ln::channelmanager::{BREAKDOWN_TIMEOUT, ChainParameters, MIN_CLTV_EXPIRY_DELTA, PaymentId};
935         use lightning::ln::features::{ChannelFeatures, NodeFeatures};
936         use lightning::ln::functional_test_utils::*;
937         use lightning::ln::msgs::{ChannelMessageHandler, Init};
938         use lightning::ln::peer_handler::{PeerManager, MessageHandler, SocketDescriptor, IgnoringMessageHandler};
939         use lightning::routing::gossip::{NetworkGraph, P2PGossipSync};
940         use lightning::routing::scoring::{ChannelUsage, ScoreUpdate, ScoreLookUp, LockableScore};
941         use lightning::routing::router::{DefaultRouter, Path, RouteHop, CandidateRouteHop};
942         use lightning::util::config::UserConfig;
943         use lightning::util::ser::Writeable;
944         use lightning::util::test_utils;
945         use lightning::util::persist::{KVStore,
946                 CHANNEL_MANAGER_PERSISTENCE_PRIMARY_NAMESPACE, CHANNEL_MANAGER_PERSISTENCE_SECONDARY_NAMESPACE, CHANNEL_MANAGER_PERSISTENCE_KEY,
947                 NETWORK_GRAPH_PERSISTENCE_PRIMARY_NAMESPACE, NETWORK_GRAPH_PERSISTENCE_SECONDARY_NAMESPACE, NETWORK_GRAPH_PERSISTENCE_KEY,
948                 SCORER_PERSISTENCE_PRIMARY_NAMESPACE, SCORER_PERSISTENCE_SECONDARY_NAMESPACE, SCORER_PERSISTENCE_KEY};
949         use lightning_persister::fs_store::FilesystemStore;
950         use std::collections::VecDeque;
951         use std::{fs, env};
952         use std::path::PathBuf;
953         use std::sync::{Arc, Mutex};
954         use std::sync::mpsc::SyncSender;
955         use std::time::Duration;
956         use lightning_rapid_gossip_sync::RapidGossipSync;
957         use super::{BackgroundProcessor, GossipSync, FRESHNESS_TIMER};
958
959         const EVENT_DEADLINE: u64 = 5 * FRESHNESS_TIMER;
960
961         #[derive(Clone, Hash, PartialEq, Eq)]
962         struct TestDescriptor{}
963         impl SocketDescriptor for TestDescriptor {
964                 fn send_data(&mut self, _data: &[u8], _resume_read: bool) -> usize {
965                         0
966                 }
967
968                 fn disconnect_socket(&mut self) {}
969         }
970
971         #[cfg(c_bindings)]
972         type LockingWrapper<T> = lightning::routing::scoring::MultiThreadedLockableScore<T>;
973         #[cfg(not(c_bindings))]
974         type LockingWrapper<T> = Mutex<T>;
975
976         type ChannelManager =
977                 channelmanager::ChannelManager<
978                         Arc<ChainMonitor>,
979                         Arc<test_utils::TestBroadcaster>,
980                         Arc<KeysManager>,
981                         Arc<KeysManager>,
982                         Arc<KeysManager>,
983                         Arc<test_utils::TestFeeEstimator>,
984                         Arc<DefaultRouter<
985                                 Arc<NetworkGraph<Arc<test_utils::TestLogger>>>,
986                                 Arc<test_utils::TestLogger>,
987                                 Arc<LockingWrapper<TestScorer>>,
988                                 (),
989                                 TestScorer>
990                         >,
991                         Arc<test_utils::TestLogger>>;
992
993         type ChainMonitor = chainmonitor::ChainMonitor<InMemorySigner, Arc<test_utils::TestChainSource>, Arc<test_utils::TestBroadcaster>, Arc<test_utils::TestFeeEstimator>, Arc<test_utils::TestLogger>, Arc<FilesystemStore>>;
994
995         type PGS = Arc<P2PGossipSync<Arc<NetworkGraph<Arc<test_utils::TestLogger>>>, Arc<test_utils::TestChainSource>, Arc<test_utils::TestLogger>>>;
996         type RGS = Arc<RapidGossipSync<Arc<NetworkGraph<Arc<test_utils::TestLogger>>>, Arc<test_utils::TestLogger>>>;
997
998         struct Node {
999                 node: Arc<ChannelManager>,
1000                 p2p_gossip_sync: PGS,
1001                 rapid_gossip_sync: RGS,
1002                 peer_manager: Arc<PeerManager<TestDescriptor, Arc<test_utils::TestChannelMessageHandler>, Arc<test_utils::TestRoutingMessageHandler>, IgnoringMessageHandler, Arc<test_utils::TestLogger>, IgnoringMessageHandler, Arc<KeysManager>>>,
1003                 chain_monitor: Arc<ChainMonitor>,
1004                 kv_store: Arc<FilesystemStore>,
1005                 tx_broadcaster: Arc<test_utils::TestBroadcaster>,
1006                 network_graph: Arc<NetworkGraph<Arc<test_utils::TestLogger>>>,
1007                 logger: Arc<test_utils::TestLogger>,
1008                 best_block: BestBlock,
1009                 scorer: Arc<LockingWrapper<TestScorer>>,
1010         }
1011
1012         impl Node {
1013                 fn p2p_gossip_sync(&self) -> GossipSync<PGS, RGS, Arc<NetworkGraph<Arc<test_utils::TestLogger>>>, Arc<test_utils::TestChainSource>, Arc<test_utils::TestLogger>> {
1014                         GossipSync::P2P(self.p2p_gossip_sync.clone())
1015                 }
1016
1017                 fn rapid_gossip_sync(&self) -> GossipSync<PGS, RGS, Arc<NetworkGraph<Arc<test_utils::TestLogger>>>, Arc<test_utils::TestChainSource>, Arc<test_utils::TestLogger>> {
1018                         GossipSync::Rapid(self.rapid_gossip_sync.clone())
1019                 }
1020
1021                 fn no_gossip_sync(&self) -> GossipSync<PGS, RGS, Arc<NetworkGraph<Arc<test_utils::TestLogger>>>, Arc<test_utils::TestChainSource>, Arc<test_utils::TestLogger>> {
1022                         GossipSync::None
1023                 }
1024         }
1025
1026         impl Drop for Node {
1027                 fn drop(&mut self) {
1028                         let data_dir = self.kv_store.get_data_dir();
1029                         match fs::remove_dir_all(data_dir.clone()) {
1030                                 Err(e) => println!("Failed to remove test store directory {}: {}", data_dir.display(), e),
1031                                 _ => {}
1032                         }
1033                 }
1034         }
1035
1036         struct Persister {
1037                 graph_error: Option<(std::io::ErrorKind, &'static str)>,
1038                 graph_persistence_notifier: Option<SyncSender<()>>,
1039                 manager_error: Option<(std::io::ErrorKind, &'static str)>,
1040                 scorer_error: Option<(std::io::ErrorKind, &'static str)>,
1041                 kv_store: FilesystemStore,
1042         }
1043
1044         impl Persister {
1045                 fn new(data_dir: PathBuf) -> Self {
1046                         let kv_store = FilesystemStore::new(data_dir);
1047                         Self { graph_error: None, graph_persistence_notifier: None, manager_error: None, scorer_error: None, kv_store }
1048                 }
1049
1050                 fn with_graph_error(self, error: std::io::ErrorKind, message: &'static str) -> Self {
1051                         Self { graph_error: Some((error, message)), ..self }
1052                 }
1053
1054                 fn with_graph_persistence_notifier(self, sender: SyncSender<()>) -> Self {
1055                         Self { graph_persistence_notifier: Some(sender), ..self }
1056                 }
1057
1058                 fn with_manager_error(self, error: std::io::ErrorKind, message: &'static str) -> Self {
1059                         Self { manager_error: Some((error, message)), ..self }
1060                 }
1061
1062                 fn with_scorer_error(self, error: std::io::ErrorKind, message: &'static str) -> Self {
1063                         Self { scorer_error: Some((error, message)), ..self }
1064                 }
1065         }
1066
1067         impl KVStore for Persister {
1068                 fn read(&self, primary_namespace: &str, secondary_namespace: &str, key: &str) -> lightning::io::Result<Vec<u8>> {
1069                         self.kv_store.read(primary_namespace, secondary_namespace, key)
1070                 }
1071
1072                 fn write(&self, primary_namespace: &str, secondary_namespace: &str, key: &str, buf: &[u8]) -> lightning::io::Result<()> {
1073                         if primary_namespace == CHANNEL_MANAGER_PERSISTENCE_PRIMARY_NAMESPACE &&
1074                                 secondary_namespace == CHANNEL_MANAGER_PERSISTENCE_SECONDARY_NAMESPACE &&
1075                                 key == CHANNEL_MANAGER_PERSISTENCE_KEY
1076                         {
1077                                 if let Some((error, message)) = self.manager_error {
1078                                         return Err(std::io::Error::new(error, message))
1079                                 }
1080                         }
1081
1082                         if primary_namespace == NETWORK_GRAPH_PERSISTENCE_PRIMARY_NAMESPACE &&
1083                                 secondary_namespace == NETWORK_GRAPH_PERSISTENCE_SECONDARY_NAMESPACE &&
1084                                 key == NETWORK_GRAPH_PERSISTENCE_KEY
1085                         {
1086                                 if let Some(sender) = &self.graph_persistence_notifier {
1087                                         match sender.send(()) {
1088                                                 Ok(()) => {},
1089                                                 Err(std::sync::mpsc::SendError(())) => println!("Persister failed to notify as receiver went away."),
1090                                         }
1091                                 };
1092
1093                                 if let Some((error, message)) = self.graph_error {
1094                                         return Err(std::io::Error::new(error, message))
1095                                 }
1096                         }
1097
1098                         if primary_namespace == SCORER_PERSISTENCE_PRIMARY_NAMESPACE &&
1099                                 secondary_namespace == SCORER_PERSISTENCE_SECONDARY_NAMESPACE &&
1100                                 key == SCORER_PERSISTENCE_KEY
1101                         {
1102                                 if let Some((error, message)) = self.scorer_error {
1103                                         return Err(std::io::Error::new(error, message))
1104                                 }
1105                         }
1106
1107                         self.kv_store.write(primary_namespace, secondary_namespace, key, buf)
1108                 }
1109
1110                 fn remove(&self, primary_namespace: &str, secondary_namespace: &str, key: &str, lazy: bool) -> lightning::io::Result<()> {
1111                         self.kv_store.remove(primary_namespace, secondary_namespace, key, lazy)
1112                 }
1113
1114                 fn list(&self, primary_namespace: &str, secondary_namespace: &str) -> lightning::io::Result<Vec<String>> {
1115                         self.kv_store.list(primary_namespace, secondary_namespace)
1116                 }
1117         }
1118
1119         struct TestScorer {
1120                 event_expectations: Option<VecDeque<TestResult>>,
1121         }
1122
1123         #[derive(Debug)]
1124         enum TestResult {
1125                 PaymentFailure { path: Path, short_channel_id: u64 },
1126                 PaymentSuccess { path: Path },
1127                 ProbeFailure { path: Path },
1128                 ProbeSuccess { path: Path },
1129         }
1130
1131         impl TestScorer {
1132                 fn new() -> Self {
1133                         Self { event_expectations: None }
1134                 }
1135
1136                 fn expect(&mut self, expectation: TestResult) {
1137                         self.event_expectations.get_or_insert_with(VecDeque::new).push_back(expectation);
1138                 }
1139         }
1140
1141         impl lightning::util::ser::Writeable for TestScorer {
1142                 fn write<W: lightning::util::ser::Writer>(&self, _: &mut W) -> Result<(), lightning::io::Error> { Ok(()) }
1143         }
1144
1145         impl ScoreLookUp for TestScorer {
1146                 type ScoreParams = ();
1147                 fn channel_penalty_msat(
1148                         &self, _candidate: &CandidateRouteHop, _usage: ChannelUsage, _score_params: &Self::ScoreParams
1149                 ) -> u64 { unimplemented!(); }
1150         }
1151
1152         impl ScoreUpdate for TestScorer {
1153                 fn payment_path_failed(&mut self, actual_path: &Path, actual_short_channel_id: u64, _: Duration) {
1154                         if let Some(expectations) = &mut self.event_expectations {
1155                                 match expectations.pop_front().unwrap() {
1156                                         TestResult::PaymentFailure { path, short_channel_id } => {
1157                                                 assert_eq!(actual_path, &path);
1158                                                 assert_eq!(actual_short_channel_id, short_channel_id);
1159                                         },
1160                                         TestResult::PaymentSuccess { path } => {
1161                                                 panic!("Unexpected successful payment path: {:?}", path)
1162                                         },
1163                                         TestResult::ProbeFailure { path } => {
1164                                                 panic!("Unexpected probe failure: {:?}", path)
1165                                         },
1166                                         TestResult::ProbeSuccess { path } => {
1167                                                 panic!("Unexpected probe success: {:?}", path)
1168                                         }
1169                                 }
1170                         }
1171                 }
1172
1173                 fn payment_path_successful(&mut self, actual_path: &Path, _: Duration) {
1174                         if let Some(expectations) = &mut self.event_expectations {
1175                                 match expectations.pop_front().unwrap() {
1176                                         TestResult::PaymentFailure { path, .. } => {
1177                                                 panic!("Unexpected payment path failure: {:?}", path)
1178                                         },
1179                                         TestResult::PaymentSuccess { path } => {
1180                                                 assert_eq!(actual_path, &path);
1181                                         },
1182                                         TestResult::ProbeFailure { path } => {
1183                                                 panic!("Unexpected probe failure: {:?}", path)
1184                                         },
1185                                         TestResult::ProbeSuccess { path } => {
1186                                                 panic!("Unexpected probe success: {:?}", path)
1187                                         }
1188                                 }
1189                         }
1190                 }
1191
1192                 fn probe_failed(&mut self, actual_path: &Path, _: u64, _: Duration) {
1193                         if let Some(expectations) = &mut self.event_expectations {
1194                                 match expectations.pop_front().unwrap() {
1195                                         TestResult::PaymentFailure { path, .. } => {
1196                                                 panic!("Unexpected payment path failure: {:?}", path)
1197                                         },
1198                                         TestResult::PaymentSuccess { path } => {
1199                                                 panic!("Unexpected payment path success: {:?}", path)
1200                                         },
1201                                         TestResult::ProbeFailure { path } => {
1202                                                 assert_eq!(actual_path, &path);
1203                                         },
1204                                         TestResult::ProbeSuccess { path } => {
1205                                                 panic!("Unexpected probe success: {:?}", path)
1206                                         }
1207                                 }
1208                         }
1209                 }
1210                 fn probe_successful(&mut self, actual_path: &Path, _: Duration) {
1211                         if let Some(expectations) = &mut self.event_expectations {
1212                                 match expectations.pop_front().unwrap() {
1213                                         TestResult::PaymentFailure { path, .. } => {
1214                                                 panic!("Unexpected payment path failure: {:?}", path)
1215                                         },
1216                                         TestResult::PaymentSuccess { path } => {
1217                                                 panic!("Unexpected payment path success: {:?}", path)
1218                                         },
1219                                         TestResult::ProbeFailure { path } => {
1220                                                 panic!("Unexpected probe failure: {:?}", path)
1221                                         },
1222                                         TestResult::ProbeSuccess { path } => {
1223                                                 assert_eq!(actual_path, &path);
1224                                         }
1225                                 }
1226                         }
1227                 }
1228                 fn time_passed(&mut self, _: Duration) {}
1229         }
1230
1231         #[cfg(c_bindings)]
1232         impl lightning::routing::scoring::Score for TestScorer {}
1233
1234         impl Drop for TestScorer {
1235                 fn drop(&mut self) {
1236                         if std::thread::panicking() {
1237                                 return;
1238                         }
1239
1240                         if let Some(event_expectations) = &self.event_expectations {
1241                                 if !event_expectations.is_empty() {
1242                                         panic!("Unsatisfied event expectations: {:?}", event_expectations);
1243                                 }
1244                         }
1245                 }
1246         }
1247
1248         fn get_full_filepath(filepath: String, filename: String) -> String {
1249                 let mut path = PathBuf::from(filepath);
1250                 path.push(filename);
1251                 path.to_str().unwrap().to_string()
1252         }
1253
1254         fn create_nodes(num_nodes: usize, persist_dir: &str) -> (String, Vec<Node>) {
1255                 let persist_temp_path = env::temp_dir().join(persist_dir);
1256                 let persist_dir = persist_temp_path.to_string_lossy().to_string();
1257                 let network = Network::Bitcoin;
1258                 let mut nodes = Vec::new();
1259                 for i in 0..num_nodes {
1260                         let tx_broadcaster = Arc::new(test_utils::TestBroadcaster::new(network));
1261                         let fee_estimator = Arc::new(test_utils::TestFeeEstimator { sat_per_kw: Mutex::new(253) });
1262                         let logger = Arc::new(test_utils::TestLogger::with_id(format!("node {}", i)));
1263                         let genesis_block = genesis_block(network);
1264                         let network_graph = Arc::new(NetworkGraph::new(network, logger.clone()));
1265                         let scorer = Arc::new(LockingWrapper::new(TestScorer::new()));
1266                         let seed = [i as u8; 32];
1267                         let router = Arc::new(DefaultRouter::new(network_graph.clone(), logger.clone(), seed, scorer.clone(), Default::default()));
1268                         let chain_source = Arc::new(test_utils::TestChainSource::new(Network::Bitcoin));
1269                         let kv_store = Arc::new(FilesystemStore::new(format!("{}_persister_{}", &persist_dir, i).into()));
1270                         let now = Duration::from_secs(genesis_block.header.time as u64);
1271                         let keys_manager = Arc::new(KeysManager::new(&seed, now.as_secs(), now.subsec_nanos()));
1272                         let chain_monitor = Arc::new(chainmonitor::ChainMonitor::new(Some(chain_source.clone()), tx_broadcaster.clone(), logger.clone(), fee_estimator.clone(), kv_store.clone()));
1273                         let best_block = BestBlock::from_network(network);
1274                         let params = ChainParameters { network, best_block };
1275                         let manager = Arc::new(ChannelManager::new(fee_estimator.clone(), chain_monitor.clone(), tx_broadcaster.clone(), router.clone(), logger.clone(), keys_manager.clone(), keys_manager.clone(), keys_manager.clone(), UserConfig::default(), params, genesis_block.header.time));
1276                         let p2p_gossip_sync = Arc::new(P2PGossipSync::new(network_graph.clone(), Some(chain_source.clone()), logger.clone()));
1277                         let rapid_gossip_sync = Arc::new(RapidGossipSync::new(network_graph.clone(), logger.clone()));
1278                         let msg_handler = MessageHandler {
1279                                 chan_handler: Arc::new(test_utils::TestChannelMessageHandler::new(ChainHash::using_genesis_block(Network::Testnet))),
1280                                 route_handler: Arc::new(test_utils::TestRoutingMessageHandler::new()),
1281                                 onion_message_handler: IgnoringMessageHandler{}, custom_message_handler: IgnoringMessageHandler{}
1282                         };
1283                         let peer_manager = Arc::new(PeerManager::new(msg_handler, 0, &seed, logger.clone(), keys_manager.clone()));
1284                         let node = Node { node: manager, p2p_gossip_sync, rapid_gossip_sync, peer_manager, chain_monitor, kv_store, tx_broadcaster, network_graph, logger, best_block, scorer };
1285                         nodes.push(node);
1286                 }
1287
1288                 for i in 0..num_nodes {
1289                         for j in (i+1)..num_nodes {
1290                                 nodes[i].node.peer_connected(&nodes[j].node.get_our_node_id(), &Init {
1291                                         features: nodes[j].node.init_features(), networks: None, remote_network_address: None
1292                                 }, true).unwrap();
1293                                 nodes[j].node.peer_connected(&nodes[i].node.get_our_node_id(), &Init {
1294                                         features: nodes[i].node.init_features(), networks: None, remote_network_address: None
1295                                 }, false).unwrap();
1296                         }
1297                 }
1298
1299                 (persist_dir, nodes)
1300         }
1301
1302         macro_rules! open_channel {
1303                 ($node_a: expr, $node_b: expr, $channel_value: expr) => {{
1304                         begin_open_channel!($node_a, $node_b, $channel_value);
1305                         let events = $node_a.node.get_and_clear_pending_events();
1306                         assert_eq!(events.len(), 1);
1307                         let (temporary_channel_id, tx) = handle_funding_generation_ready!(events[0], $channel_value);
1308                         $node_a.node.funding_transaction_generated(&temporary_channel_id, &$node_b.node.get_our_node_id(), tx.clone()).unwrap();
1309                         $node_b.node.handle_funding_created(&$node_a.node.get_our_node_id(), &get_event_msg!($node_a, MessageSendEvent::SendFundingCreated, $node_b.node.get_our_node_id()));
1310                         get_event!($node_b, Event::ChannelPending);
1311                         $node_a.node.handle_funding_signed(&$node_b.node.get_our_node_id(), &get_event_msg!($node_b, MessageSendEvent::SendFundingSigned, $node_a.node.get_our_node_id()));
1312                         get_event!($node_a, Event::ChannelPending);
1313                         tx
1314                 }}
1315         }
1316
1317         macro_rules! begin_open_channel {
1318                 ($node_a: expr, $node_b: expr, $channel_value: expr) => {{
1319                         $node_a.node.create_channel($node_b.node.get_our_node_id(), $channel_value, 100, 42, None, None).unwrap();
1320                         $node_b.node.handle_open_channel(&$node_a.node.get_our_node_id(), &get_event_msg!($node_a, MessageSendEvent::SendOpenChannel, $node_b.node.get_our_node_id()));
1321                         $node_a.node.handle_accept_channel(&$node_b.node.get_our_node_id(), &get_event_msg!($node_b, MessageSendEvent::SendAcceptChannel, $node_a.node.get_our_node_id()));
1322                 }}
1323         }
1324
1325         macro_rules! handle_funding_generation_ready {
1326                 ($event: expr, $channel_value: expr) => {{
1327                         match $event {
1328                                 Event::FundingGenerationReady { temporary_channel_id, channel_value_satoshis, ref output_script, user_channel_id, .. } => {
1329                                         assert_eq!(channel_value_satoshis, $channel_value);
1330                                         assert_eq!(user_channel_id, 42);
1331
1332                                         let tx = Transaction { version: 1 as i32, lock_time: LockTime::ZERO, input: Vec::new(), output: vec![TxOut {
1333                                                 value: channel_value_satoshis, script_pubkey: output_script.clone(),
1334                                         }]};
1335                                         (temporary_channel_id, tx)
1336                                 },
1337                                 _ => panic!("Unexpected event"),
1338                         }
1339                 }}
1340         }
1341
1342         fn confirm_transaction_depth(node: &mut Node, tx: &Transaction, depth: u32) {
1343                 for i in 1..=depth {
1344                         let prev_blockhash = node.best_block.block_hash();
1345                         let height = node.best_block.height() + 1;
1346                         let header = create_dummy_header(prev_blockhash, height);
1347                         let txdata = vec![(0, tx)];
1348                         node.best_block = BestBlock::new(header.block_hash(), height);
1349                         match i {
1350                                 1 => {
1351                                         node.node.transactions_confirmed(&header, &txdata, height);
1352                                         node.chain_monitor.transactions_confirmed(&header, &txdata, height);
1353                                 },
1354                                 x if x == depth => {
1355                                         node.node.best_block_updated(&header, height);
1356                                         node.chain_monitor.best_block_updated(&header, height);
1357                                 },
1358                                 _ => {},
1359                         }
1360                 }
1361         }
1362         fn confirm_transaction(node: &mut Node, tx: &Transaction) {
1363                 confirm_transaction_depth(node, tx, ANTI_REORG_DELAY);
1364         }
1365
1366         #[test]
1367         fn test_background_processor() {
1368                 // Test that when a new channel is created, the ChannelManager needs to be re-persisted with
1369                 // updates. Also test that when new updates are available, the manager signals that it needs
1370                 // re-persistence and is successfully re-persisted.
1371                 let (persist_dir, nodes) = create_nodes(2, "test_background_processor");
1372
1373                 // Go through the channel creation process so that each node has something to persist. Since
1374                 // open_channel consumes events, it must complete before starting BackgroundProcessor to
1375                 // avoid a race with processing events.
1376                 let tx = open_channel!(nodes[0], nodes[1], 100000);
1377
1378                 // Initiate the background processors to watch each node.
1379                 let data_dir = nodes[0].kv_store.get_data_dir();
1380                 let persister = Arc::new(Persister::new(data_dir));
1381                 let event_handler = |_: _| {};
1382                 let bg_processor = BackgroundProcessor::start(persister, event_handler, nodes[0].chain_monitor.clone(), nodes[0].node.clone(), nodes[0].p2p_gossip_sync(), nodes[0].peer_manager.clone(), nodes[0].logger.clone(), Some(nodes[0].scorer.clone()));
1383
1384                 macro_rules! check_persisted_data {
1385                         ($node: expr, $filepath: expr) => {
1386                                 let mut expected_bytes = Vec::new();
1387                                 loop {
1388                                         expected_bytes.clear();
1389                                         match $node.write(&mut expected_bytes) {
1390                                                 Ok(()) => {
1391                                                         match std::fs::read($filepath) {
1392                                                                 Ok(bytes) => {
1393                                                                         if bytes == expected_bytes {
1394                                                                                 break
1395                                                                         } else {
1396                                                                                 continue
1397                                                                         }
1398                                                                 },
1399                                                                 Err(_) => continue
1400                                                         }
1401                                                 },
1402                                                 Err(e) => panic!("Unexpected error: {}", e)
1403                                         }
1404                                 }
1405                         }
1406                 }
1407
1408                 // Check that the initial channel manager data is persisted as expected.
1409                 let filepath = get_full_filepath(format!("{}_persister_0", &persist_dir), "manager".to_string());
1410                 check_persisted_data!(nodes[0].node, filepath.clone());
1411
1412                 loop {
1413                         if !nodes[0].node.get_event_or_persist_condvar_value() { break }
1414                 }
1415
1416                 // Force-close the channel.
1417                 nodes[0].node.force_close_broadcasting_latest_txn(&OutPoint { txid: tx.txid(), index: 0 }.to_channel_id(), &nodes[1].node.get_our_node_id()).unwrap();
1418
1419                 // Check that the force-close updates are persisted.
1420                 check_persisted_data!(nodes[0].node, filepath.clone());
1421                 loop {
1422                         if !nodes[0].node.get_event_or_persist_condvar_value() { break }
1423                 }
1424
1425                 // Check network graph is persisted
1426                 let filepath = get_full_filepath(format!("{}_persister_0", &persist_dir), "network_graph".to_string());
1427                 check_persisted_data!(nodes[0].network_graph, filepath.clone());
1428
1429                 // Check scorer is persisted
1430                 let filepath = get_full_filepath(format!("{}_persister_0", &persist_dir), "scorer".to_string());
1431                 check_persisted_data!(nodes[0].scorer, filepath.clone());
1432
1433                 if !std::thread::panicking() {
1434                         bg_processor.stop().unwrap();
1435                 }
1436         }
1437
1438         #[test]
1439         fn test_timer_tick_called() {
1440                 // Test that:
1441                 // - `ChannelManager::timer_tick_occurred` is called every `FRESHNESS_TIMER`,
1442                 // - `ChainMonitor::rebroadcast_pending_claims` is called every `REBROADCAST_TIMER`,
1443                 // - `PeerManager::timer_tick_occurred` is called every `PING_TIMER`, and
1444                 // - `OnionMessageHandler::timer_tick_occurred` is called every `ONION_MESSAGE_HANDLER_TIMER`.
1445                 let (_, nodes) = create_nodes(1, "test_timer_tick_called");
1446                 let data_dir = nodes[0].kv_store.get_data_dir();
1447                 let persister = Arc::new(Persister::new(data_dir));
1448                 let event_handler = |_: _| {};
1449                 let bg_processor = BackgroundProcessor::start(persister, event_handler, nodes[0].chain_monitor.clone(), nodes[0].node.clone(), nodes[0].no_gossip_sync(), nodes[0].peer_manager.clone(), nodes[0].logger.clone(), Some(nodes[0].scorer.clone()));
1450                 loop {
1451                         let log_entries = nodes[0].logger.lines.lock().unwrap();
1452                         let desired_log_1 = "Calling ChannelManager's timer_tick_occurred".to_string();
1453                         let desired_log_2 = "Calling PeerManager's timer_tick_occurred".to_string();
1454                         let desired_log_3 = "Rebroadcasting monitor's pending claims".to_string();
1455                         let desired_log_4 = "Calling OnionMessageHandler's timer_tick_occurred".to_string();
1456                         if log_entries.get(&("lightning_background_processor", desired_log_1)).is_some() &&
1457                                 log_entries.get(&("lightning_background_processor", desired_log_2)).is_some() &&
1458                                 log_entries.get(&("lightning_background_processor", desired_log_3)).is_some() &&
1459                                 log_entries.get(&("lightning_background_processor", desired_log_4)).is_some() {
1460                                 break
1461                         }
1462                 }
1463
1464                 if !std::thread::panicking() {
1465                         bg_processor.stop().unwrap();
1466                 }
1467         }
1468
1469         #[test]
1470         fn test_channel_manager_persist_error() {
1471                 // Test that if we encounter an error during manager persistence, the thread panics.
1472                 let (_, nodes) = create_nodes(2, "test_persist_error");
1473                 open_channel!(nodes[0], nodes[1], 100000);
1474
1475                 let data_dir = nodes[0].kv_store.get_data_dir();
1476                 let persister = Arc::new(Persister::new(data_dir).with_manager_error(std::io::ErrorKind::Other, "test"));
1477                 let event_handler = |_: _| {};
1478                 let bg_processor = BackgroundProcessor::start(persister, event_handler, nodes[0].chain_monitor.clone(), nodes[0].node.clone(), nodes[0].no_gossip_sync(), nodes[0].peer_manager.clone(), nodes[0].logger.clone(), Some(nodes[0].scorer.clone()));
1479                 match bg_processor.join() {
1480                         Ok(_) => panic!("Expected error persisting manager"),
1481                         Err(e) => {
1482                                 assert_eq!(e.kind(), std::io::ErrorKind::Other);
1483                                 assert_eq!(e.get_ref().unwrap().to_string(), "test");
1484                         },
1485                 }
1486         }
1487
1488         #[tokio::test]
1489         #[cfg(feature = "futures")]
1490         async fn test_channel_manager_persist_error_async() {
1491                 // Test that if we encounter an error during manager persistence, the thread panics.
1492                 let (_, nodes) = create_nodes(2, "test_persist_error_sync");
1493                 open_channel!(nodes[0], nodes[1], 100000);
1494
1495                 let data_dir = nodes[0].kv_store.get_data_dir();
1496                 let persister = Arc::new(Persister::new(data_dir).with_manager_error(std::io::ErrorKind::Other, "test"));
1497
1498                 let bp_future = super::process_events_async(
1499                         persister, |_: _| {async {}}, nodes[0].chain_monitor.clone(), nodes[0].node.clone(),
1500                         nodes[0].rapid_gossip_sync(), nodes[0].peer_manager.clone(), nodes[0].logger.clone(),
1501                         Some(nodes[0].scorer.clone()), move |dur: Duration| {
1502                                 Box::pin(async move {
1503                                         tokio::time::sleep(dur).await;
1504                                         false // Never exit
1505                                 })
1506                         }, false, || Some(Duration::ZERO),
1507                 );
1508                 match bp_future.await {
1509                         Ok(_) => panic!("Expected error persisting manager"),
1510                         Err(e) => {
1511                                 assert_eq!(e.kind(), std::io::ErrorKind::Other);
1512                                 assert_eq!(e.get_ref().unwrap().to_string(), "test");
1513                         },
1514                 }
1515         }
1516
1517         #[test]
1518         fn test_network_graph_persist_error() {
1519                 // Test that if we encounter an error during network graph persistence, an error gets returned.
1520                 let (_, nodes) = create_nodes(2, "test_persist_network_graph_error");
1521                 let data_dir = nodes[0].kv_store.get_data_dir();
1522                 let persister = Arc::new(Persister::new(data_dir).with_graph_error(std::io::ErrorKind::Other, "test"));
1523                 let event_handler = |_: _| {};
1524                 let bg_processor = BackgroundProcessor::start(persister, event_handler, nodes[0].chain_monitor.clone(), nodes[0].node.clone(), nodes[0].p2p_gossip_sync(), nodes[0].peer_manager.clone(), nodes[0].logger.clone(), Some(nodes[0].scorer.clone()));
1525
1526                 match bg_processor.stop() {
1527                         Ok(_) => panic!("Expected error persisting network graph"),
1528                         Err(e) => {
1529                                 assert_eq!(e.kind(), std::io::ErrorKind::Other);
1530                                 assert_eq!(e.get_ref().unwrap().to_string(), "test");
1531                         },
1532                 }
1533         }
1534
1535         #[test]
1536         fn test_scorer_persist_error() {
1537                 // Test that if we encounter an error during scorer persistence, an error gets returned.
1538                 let (_, nodes) = create_nodes(2, "test_persist_scorer_error");
1539                 let data_dir = nodes[0].kv_store.get_data_dir();
1540                 let persister = Arc::new(Persister::new(data_dir).with_scorer_error(std::io::ErrorKind::Other, "test"));
1541                 let event_handler = |_: _| {};
1542                 let bg_processor = BackgroundProcessor::start(persister, event_handler, nodes[0].chain_monitor.clone(), nodes[0].node.clone(), nodes[0].no_gossip_sync(), nodes[0].peer_manager.clone(),  nodes[0].logger.clone(), Some(nodes[0].scorer.clone()));
1543
1544                 match bg_processor.stop() {
1545                         Ok(_) => panic!("Expected error persisting scorer"),
1546                         Err(e) => {
1547                                 assert_eq!(e.kind(), std::io::ErrorKind::Other);
1548                                 assert_eq!(e.get_ref().unwrap().to_string(), "test");
1549                         },
1550                 }
1551         }
1552
1553         #[test]
1554         fn test_background_event_handling() {
1555                 let (_, mut nodes) = create_nodes(2, "test_background_event_handling");
1556                 let channel_value = 100000;
1557                 let data_dir = nodes[0].kv_store.get_data_dir();
1558                 let persister = Arc::new(Persister::new(data_dir.clone()));
1559
1560                 // Set up a background event handler for FundingGenerationReady events.
1561                 let (funding_generation_send, funding_generation_recv) = std::sync::mpsc::sync_channel(1);
1562                 let (channel_pending_send, channel_pending_recv) = std::sync::mpsc::sync_channel(1);
1563                 let event_handler = move |event: Event| match event {
1564                         Event::FundingGenerationReady { .. } => funding_generation_send.send(handle_funding_generation_ready!(event, channel_value)).unwrap(),
1565                         Event::ChannelPending { .. } => channel_pending_send.send(()).unwrap(),
1566                         Event::ChannelReady { .. } => {},
1567                         _ => panic!("Unexpected event: {:?}", event),
1568                 };
1569
1570                 let bg_processor = BackgroundProcessor::start(persister, event_handler, nodes[0].chain_monitor.clone(), nodes[0].node.clone(), nodes[0].no_gossip_sync(), nodes[0].peer_manager.clone(), nodes[0].logger.clone(), Some(nodes[0].scorer.clone()));
1571
1572                 // Open a channel and check that the FundingGenerationReady event was handled.
1573                 begin_open_channel!(nodes[0], nodes[1], channel_value);
1574                 let (temporary_channel_id, funding_tx) = funding_generation_recv
1575                         .recv_timeout(Duration::from_secs(EVENT_DEADLINE))
1576                         .expect("FundingGenerationReady not handled within deadline");
1577                 nodes[0].node.funding_transaction_generated(&temporary_channel_id, &nodes[1].node.get_our_node_id(), funding_tx.clone()).unwrap();
1578                 nodes[1].node.handle_funding_created(&nodes[0].node.get_our_node_id(), &get_event_msg!(nodes[0], MessageSendEvent::SendFundingCreated, nodes[1].node.get_our_node_id()));
1579                 get_event!(nodes[1], Event::ChannelPending);
1580                 nodes[0].node.handle_funding_signed(&nodes[1].node.get_our_node_id(), &get_event_msg!(nodes[1], MessageSendEvent::SendFundingSigned, nodes[0].node.get_our_node_id()));
1581                 let _ = channel_pending_recv.recv_timeout(Duration::from_secs(EVENT_DEADLINE))
1582                         .expect("ChannelPending not handled within deadline");
1583
1584                 // Confirm the funding transaction.
1585                 confirm_transaction(&mut nodes[0], &funding_tx);
1586                 let as_funding = get_event_msg!(nodes[0], MessageSendEvent::SendChannelReady, nodes[1].node.get_our_node_id());
1587                 confirm_transaction(&mut nodes[1], &funding_tx);
1588                 let bs_funding = get_event_msg!(nodes[1], MessageSendEvent::SendChannelReady, nodes[0].node.get_our_node_id());
1589                 nodes[0].node.handle_channel_ready(&nodes[1].node.get_our_node_id(), &bs_funding);
1590                 let _as_channel_update = get_event_msg!(nodes[0], MessageSendEvent::SendChannelUpdate, nodes[1].node.get_our_node_id());
1591                 nodes[1].node.handle_channel_ready(&nodes[0].node.get_our_node_id(), &as_funding);
1592                 let _bs_channel_update = get_event_msg!(nodes[1], MessageSendEvent::SendChannelUpdate, nodes[0].node.get_our_node_id());
1593
1594                 if !std::thread::panicking() {
1595                         bg_processor.stop().unwrap();
1596                 }
1597
1598                 // Set up a background event handler for SpendableOutputs events.
1599                 let (sender, receiver) = std::sync::mpsc::sync_channel(1);
1600                 let event_handler = move |event: Event| match event {
1601                         Event::SpendableOutputs { .. } => sender.send(event).unwrap(),
1602                         Event::ChannelReady { .. } => {},
1603                         Event::ChannelClosed { .. } => {},
1604                         _ => panic!("Unexpected event: {:?}", event),
1605                 };
1606                 let persister = Arc::new(Persister::new(data_dir));
1607                 let bg_processor = BackgroundProcessor::start(persister, event_handler, nodes[0].chain_monitor.clone(), nodes[0].node.clone(), nodes[0].no_gossip_sync(), nodes[0].peer_manager.clone(), nodes[0].logger.clone(), Some(nodes[0].scorer.clone()));
1608
1609                 // Force close the channel and check that the SpendableOutputs event was handled.
1610                 nodes[0].node.force_close_broadcasting_latest_txn(&nodes[0].node.list_channels()[0].channel_id, &nodes[1].node.get_our_node_id()).unwrap();
1611                 let commitment_tx = nodes[0].tx_broadcaster.txn_broadcasted.lock().unwrap().pop().unwrap();
1612                 confirm_transaction_depth(&mut nodes[0], &commitment_tx, BREAKDOWN_TIMEOUT as u32);
1613
1614                 let event = receiver
1615                         .recv_timeout(Duration::from_secs(EVENT_DEADLINE))
1616                         .expect("Events not handled within deadline");
1617                 match event {
1618                         Event::SpendableOutputs { .. } => {},
1619                         _ => panic!("Unexpected event: {:?}", event),
1620                 }
1621
1622                 if !std::thread::panicking() {
1623                         bg_processor.stop().unwrap();
1624                 }
1625         }
1626
1627         #[test]
1628         fn test_scorer_persistence() {
1629                 let (_, nodes) = create_nodes(2, "test_scorer_persistence");
1630                 let data_dir = nodes[0].kv_store.get_data_dir();
1631                 let persister = Arc::new(Persister::new(data_dir));
1632                 let event_handler = |_: _| {};
1633                 let bg_processor = BackgroundProcessor::start(persister, event_handler, nodes[0].chain_monitor.clone(), nodes[0].node.clone(), nodes[0].no_gossip_sync(), nodes[0].peer_manager.clone(), nodes[0].logger.clone(), Some(nodes[0].scorer.clone()));
1634
1635                 loop {
1636                         let log_entries = nodes[0].logger.lines.lock().unwrap();
1637                         let expected_log = "Calling time_passed and persisting scorer".to_string();
1638                         if log_entries.get(&("lightning_background_processor", expected_log)).is_some() {
1639                                 break
1640                         }
1641                 }
1642
1643                 if !std::thread::panicking() {
1644                         bg_processor.stop().unwrap();
1645                 }
1646         }
1647
1648         macro_rules! do_test_not_pruning_network_graph_until_graph_sync_completion {
1649                 ($nodes: expr, $receive: expr, $sleep: expr) => {
1650                         let features = ChannelFeatures::empty();
1651                         $nodes[0].network_graph.add_channel_from_partial_announcement(
1652                                 42, 53, features, $nodes[0].node.get_our_node_id(), $nodes[1].node.get_our_node_id()
1653                         ).expect("Failed to update channel from partial announcement");
1654                         let original_graph_description = $nodes[0].network_graph.to_string();
1655                         assert!(original_graph_description.contains("42: features: 0000, node_one:"));
1656                         assert_eq!($nodes[0].network_graph.read_only().channels().len(), 1);
1657
1658                         loop {
1659                                 $sleep;
1660                                 let log_entries = $nodes[0].logger.lines.lock().unwrap();
1661                                 let loop_counter = "Calling ChannelManager's timer_tick_occurred".to_string();
1662                                 if *log_entries.get(&("lightning_background_processor", loop_counter))
1663                                         .unwrap_or(&0) > 1
1664                                 {
1665                                         // Wait until the loop has gone around at least twice.
1666                                         break
1667                                 }
1668                         }
1669
1670                         let initialization_input = vec![
1671                                 76, 68, 75, 1, 111, 226, 140, 10, 182, 241, 179, 114, 193, 166, 162, 70, 174, 99, 247,
1672                                 79, 147, 30, 131, 101, 225, 90, 8, 156, 104, 214, 25, 0, 0, 0, 0, 0, 97, 227, 98, 218,
1673                                 0, 0, 0, 4, 2, 22, 7, 207, 206, 25, 164, 197, 231, 230, 231, 56, 102, 61, 250, 251,
1674                                 187, 172, 38, 46, 79, 247, 108, 44, 155, 48, 219, 238, 252, 53, 192, 6, 67, 2, 36, 125,
1675                                 157, 176, 223, 175, 234, 116, 94, 248, 201, 225, 97, 235, 50, 47, 115, 172, 63, 136,
1676                                 88, 216, 115, 11, 111, 217, 114, 84, 116, 124, 231, 107, 2, 158, 1, 242, 121, 152, 106,
1677                                 204, 131, 186, 35, 93, 70, 216, 10, 237, 224, 183, 89, 95, 65, 3, 83, 185, 58, 138,
1678                                 181, 64, 187, 103, 127, 68, 50, 2, 201, 19, 17, 138, 136, 149, 185, 226, 156, 137, 175,
1679                                 110, 32, 237, 0, 217, 90, 31, 100, 228, 149, 46, 219, 175, 168, 77, 4, 143, 38, 128,
1680                                 76, 97, 0, 0, 0, 2, 0, 0, 255, 8, 153, 192, 0, 2, 27, 0, 0, 0, 1, 0, 0, 255, 2, 68,
1681                                 226, 0, 6, 11, 0, 1, 2, 3, 0, 0, 0, 2, 0, 40, 0, 0, 0, 0, 0, 0, 3, 232, 0, 0, 3, 232,
1682                                 0, 0, 0, 1, 0, 0, 0, 0, 58, 85, 116, 216, 255, 8, 153, 192, 0, 2, 27, 0, 0, 25, 0, 0,
1683                                 0, 1, 0, 0, 0, 125, 255, 2, 68, 226, 0, 6, 11, 0, 1, 5, 0, 0, 0, 0, 29, 129, 25, 192,
1684                         ];
1685                         $nodes[0].rapid_gossip_sync.update_network_graph_no_std(&initialization_input[..], Some(1642291930)).unwrap();
1686
1687                         // this should have added two channels and pruned the previous one.
1688                         assert_eq!($nodes[0].network_graph.read_only().channels().len(), 2);
1689
1690                         $receive.expect("Network graph not pruned within deadline");
1691
1692                         // all channels should now be pruned
1693                         assert_eq!($nodes[0].network_graph.read_only().channels().len(), 0);
1694                 }
1695         }
1696
1697         #[test]
1698         fn test_not_pruning_network_graph_until_graph_sync_completion() {
1699                 let (sender, receiver) = std::sync::mpsc::sync_channel(1);
1700
1701                 let (_, nodes) = create_nodes(2, "test_not_pruning_network_graph_until_graph_sync_completion");
1702                 let data_dir = nodes[0].kv_store.get_data_dir();
1703                 let persister = Arc::new(Persister::new(data_dir).with_graph_persistence_notifier(sender));
1704
1705                 let event_handler = |_: _| {};
1706                 let background_processor = BackgroundProcessor::start(persister, event_handler, nodes[0].chain_monitor.clone(), nodes[0].node.clone(), nodes[0].rapid_gossip_sync(), nodes[0].peer_manager.clone(), nodes[0].logger.clone(), Some(nodes[0].scorer.clone()));
1707
1708                 do_test_not_pruning_network_graph_until_graph_sync_completion!(nodes,
1709                         receiver.recv_timeout(Duration::from_secs(super::FIRST_NETWORK_PRUNE_TIMER * 5)),
1710                         std::thread::sleep(Duration::from_millis(1)));
1711
1712                 background_processor.stop().unwrap();
1713         }
1714
1715         #[tokio::test]
1716         #[cfg(feature = "futures")]
1717         async fn test_not_pruning_network_graph_until_graph_sync_completion_async() {
1718                 let (sender, receiver) = std::sync::mpsc::sync_channel(1);
1719
1720                 let (_, nodes) = create_nodes(2, "test_not_pruning_network_graph_until_graph_sync_completion_async");
1721                 let data_dir = nodes[0].kv_store.get_data_dir();
1722                 let persister = Arc::new(Persister::new(data_dir).with_graph_persistence_notifier(sender));
1723
1724                 let (exit_sender, exit_receiver) = tokio::sync::watch::channel(());
1725                 let bp_future = super::process_events_async(
1726                         persister, |_: _| {async {}}, nodes[0].chain_monitor.clone(), nodes[0].node.clone(),
1727                         nodes[0].rapid_gossip_sync(), nodes[0].peer_manager.clone(), nodes[0].logger.clone(),
1728                         Some(nodes[0].scorer.clone()), move |dur: Duration| {
1729                                 let mut exit_receiver = exit_receiver.clone();
1730                                 Box::pin(async move {
1731                                         tokio::select! {
1732                                                 _ = tokio::time::sleep(dur) => false,
1733                                                 _ = exit_receiver.changed() => true,
1734                                         }
1735                                 })
1736                         }, false, || Some(Duration::from_secs(1696300000)),
1737                 );
1738
1739                 let t1 = tokio::spawn(bp_future);
1740                 let t2 = tokio::spawn(async move {
1741                         do_test_not_pruning_network_graph_until_graph_sync_completion!(nodes, {
1742                                 let mut i = 0;
1743                                 loop {
1744                                         tokio::time::sleep(Duration::from_secs(super::FIRST_NETWORK_PRUNE_TIMER)).await;
1745                                         if let Ok(()) = receiver.try_recv() { break Ok::<(), ()>(()); }
1746                                         assert!(i < 5);
1747                                         i += 1;
1748                                 }
1749                         }, tokio::time::sleep(Duration::from_millis(1)).await);
1750                         exit_sender.send(()).unwrap();
1751                 });
1752                 let (r1, r2) = tokio::join!(t1, t2);
1753                 r1.unwrap().unwrap();
1754                 r2.unwrap()
1755         }
1756
1757         macro_rules! do_test_payment_path_scoring {
1758                 ($nodes: expr, $receive: expr) => {
1759                         // Ensure that we update the scorer when relevant events are processed. In this case, we ensure
1760                         // that we update the scorer upon a payment path succeeding (note that the channel must be
1761                         // public or else we won't score it).
1762                         // A background event handler for FundingGenerationReady events must be hooked up to a
1763                         // running background processor.
1764                         let scored_scid = 4242;
1765                         let secp_ctx = Secp256k1::new();
1766                         let node_1_privkey = SecretKey::from_slice(&[42; 32]).unwrap();
1767                         let node_1_id = PublicKey::from_secret_key(&secp_ctx, &node_1_privkey);
1768
1769                         let path = Path { hops: vec![RouteHop {
1770                                 pubkey: node_1_id,
1771                                 node_features: NodeFeatures::empty(),
1772                                 short_channel_id: scored_scid,
1773                                 channel_features: ChannelFeatures::empty(),
1774                                 fee_msat: 0,
1775                                 cltv_expiry_delta: MIN_CLTV_EXPIRY_DELTA as u32,
1776                                 maybe_announced_channel: true,
1777                         }], blinded_tail: None };
1778
1779                         $nodes[0].scorer.write_lock().expect(TestResult::PaymentFailure { path: path.clone(), short_channel_id: scored_scid });
1780                         $nodes[0].node.push_pending_event(Event::PaymentPathFailed {
1781                                 payment_id: None,
1782                                 payment_hash: PaymentHash([42; 32]),
1783                                 payment_failed_permanently: false,
1784                                 failure: PathFailure::OnPath { network_update: None },
1785                                 path: path.clone(),
1786                                 short_channel_id: Some(scored_scid),
1787                         });
1788                         let event = $receive.expect("PaymentPathFailed not handled within deadline");
1789                         match event {
1790                                 Event::PaymentPathFailed { .. } => {},
1791                                 _ => panic!("Unexpected event"),
1792                         }
1793
1794                         // Ensure we'll score payments that were explicitly failed back by the destination as
1795                         // ProbeSuccess.
1796                         $nodes[0].scorer.write_lock().expect(TestResult::ProbeSuccess { path: path.clone() });
1797                         $nodes[0].node.push_pending_event(Event::PaymentPathFailed {
1798                                 payment_id: None,
1799                                 payment_hash: PaymentHash([42; 32]),
1800                                 payment_failed_permanently: true,
1801                                 failure: PathFailure::OnPath { network_update: None },
1802                                 path: path.clone(),
1803                                 short_channel_id: None,
1804                         });
1805                         let event = $receive.expect("PaymentPathFailed not handled within deadline");
1806                         match event {
1807                                 Event::PaymentPathFailed { .. } => {},
1808                                 _ => panic!("Unexpected event"),
1809                         }
1810
1811                         $nodes[0].scorer.write_lock().expect(TestResult::PaymentSuccess { path: path.clone() });
1812                         $nodes[0].node.push_pending_event(Event::PaymentPathSuccessful {
1813                                 payment_id: PaymentId([42; 32]),
1814                                 payment_hash: None,
1815                                 path: path.clone(),
1816                         });
1817                         let event = $receive.expect("PaymentPathSuccessful not handled within deadline");
1818                         match event {
1819                                 Event::PaymentPathSuccessful { .. } => {},
1820                                 _ => panic!("Unexpected event"),
1821                         }
1822
1823                         $nodes[0].scorer.write_lock().expect(TestResult::ProbeSuccess { path: path.clone() });
1824                         $nodes[0].node.push_pending_event(Event::ProbeSuccessful {
1825                                 payment_id: PaymentId([42; 32]),
1826                                 payment_hash: PaymentHash([42; 32]),
1827                                 path: path.clone(),
1828                         });
1829                         let event = $receive.expect("ProbeSuccessful not handled within deadline");
1830                         match event {
1831                                 Event::ProbeSuccessful  { .. } => {},
1832                                 _ => panic!("Unexpected event"),
1833                         }
1834
1835                         $nodes[0].scorer.write_lock().expect(TestResult::ProbeFailure { path: path.clone() });
1836                         $nodes[0].node.push_pending_event(Event::ProbeFailed {
1837                                 payment_id: PaymentId([42; 32]),
1838                                 payment_hash: PaymentHash([42; 32]),
1839                                 path,
1840                                 short_channel_id: Some(scored_scid),
1841                         });
1842                         let event = $receive.expect("ProbeFailure not handled within deadline");
1843                         match event {
1844                                 Event::ProbeFailed { .. } => {},
1845                                 _ => panic!("Unexpected event"),
1846                         }
1847                 }
1848         }
1849
1850         #[test]
1851         fn test_payment_path_scoring() {
1852                 let (sender, receiver) = std::sync::mpsc::sync_channel(1);
1853                 let event_handler = move |event: Event| match event {
1854                         Event::PaymentPathFailed { .. } => sender.send(event).unwrap(),
1855                         Event::PaymentPathSuccessful { .. } => sender.send(event).unwrap(),
1856                         Event::ProbeSuccessful { .. } => sender.send(event).unwrap(),
1857                         Event::ProbeFailed { .. } => sender.send(event).unwrap(),
1858                         _ => panic!("Unexpected event: {:?}", event),
1859                 };
1860
1861                 let (_, nodes) = create_nodes(1, "test_payment_path_scoring");
1862                 let data_dir = nodes[0].kv_store.get_data_dir();
1863                 let persister = Arc::new(Persister::new(data_dir));
1864                 let bg_processor = BackgroundProcessor::start(persister, event_handler, nodes[0].chain_monitor.clone(), nodes[0].node.clone(), nodes[0].no_gossip_sync(), nodes[0].peer_manager.clone(), nodes[0].logger.clone(), Some(nodes[0].scorer.clone()));
1865
1866                 do_test_payment_path_scoring!(nodes, receiver.recv_timeout(Duration::from_secs(EVENT_DEADLINE)));
1867
1868                 if !std::thread::panicking() {
1869                         bg_processor.stop().unwrap();
1870                 }
1871
1872                 let log_entries = nodes[0].logger.lines.lock().unwrap();
1873                 let expected_log = "Persisting scorer after update".to_string();
1874                 assert_eq!(*log_entries.get(&("lightning_background_processor", expected_log)).unwrap(), 5);
1875         }
1876
1877         #[tokio::test]
1878         #[cfg(feature = "futures")]
1879         async fn test_payment_path_scoring_async() {
1880                 let (sender, mut receiver) = tokio::sync::mpsc::channel(1);
1881                 let event_handler = move |event: Event| {
1882                         let sender_ref = sender.clone();
1883                         async move {
1884                                 match event {
1885                                         Event::PaymentPathFailed { .. } => { sender_ref.send(event).await.unwrap() },
1886                                         Event::PaymentPathSuccessful { .. } => { sender_ref.send(event).await.unwrap() },
1887                                         Event::ProbeSuccessful { .. } => { sender_ref.send(event).await.unwrap() },
1888                                         Event::ProbeFailed { .. } => { sender_ref.send(event).await.unwrap() },
1889                                         _ => panic!("Unexpected event: {:?}", event),
1890                                 }
1891                         }
1892                 };
1893
1894                 let (_, nodes) = create_nodes(1, "test_payment_path_scoring_async");
1895                 let data_dir = nodes[0].kv_store.get_data_dir();
1896                 let persister = Arc::new(Persister::new(data_dir));
1897
1898                 let (exit_sender, exit_receiver) = tokio::sync::watch::channel(());
1899
1900                 let bp_future = super::process_events_async(
1901                         persister, event_handler, nodes[0].chain_monitor.clone(), nodes[0].node.clone(),
1902                         nodes[0].no_gossip_sync(), nodes[0].peer_manager.clone(), nodes[0].logger.clone(),
1903                         Some(nodes[0].scorer.clone()), move |dur: Duration| {
1904                                 let mut exit_receiver = exit_receiver.clone();
1905                                 Box::pin(async move {
1906                                         tokio::select! {
1907                                                 _ = tokio::time::sleep(dur) => false,
1908                                                 _ = exit_receiver.changed() => true,
1909                                         }
1910                                 })
1911                         }, false, || Some(Duration::ZERO),
1912                 );
1913                 let t1 = tokio::spawn(bp_future);
1914                 let t2 = tokio::spawn(async move {
1915                         do_test_payment_path_scoring!(nodes, receiver.recv().await);
1916                         exit_sender.send(()).unwrap();
1917
1918                         let log_entries = nodes[0].logger.lines.lock().unwrap();
1919                         let expected_log = "Persisting scorer after update".to_string();
1920                         assert_eq!(*log_entries.get(&("lightning_background_processor", expected_log)).unwrap(), 5);
1921                 });
1922
1923                 let (r1, r2) = tokio::join!(t1, t2);
1924                 r1.unwrap().unwrap();
1925                 r2.unwrap()
1926         }
1927 }