_ Git - rust-lightning/blob - lightning-background-processor/src/lib.rs

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