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