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