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

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