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.
5 // Prefix these with `rustdoc::` when we update our MSRV to be >= 1.52 to remove warnings.
6 #![deny(broken_intra_doc_links)]
7 #![deny(private_intra_doc_links)]
10 #![cfg_attr(not(feature = "futures"), deny(unsafe_code))]
12 #![cfg_attr(docsrs, feature(doc_auto_cfg))]
14 #![cfg_attr(all(not(feature = "std"), not(test)), no_std)]
16 #[cfg(any(test, feature = "std"))]
19 #[cfg(not(feature = "std"))]
22 #[macro_use] extern crate lightning;
23 extern crate lightning_rapid_gossip_sync;
26 use lightning::chain::chaininterface::{BroadcasterInterface, FeeEstimator};
27 use lightning::chain::chainmonitor::{ChainMonitor, Persist};
28 use lightning::sign::{EntropySource, NodeSigner, SignerProvider};
29 use lightning::events::{Event, PathFailure};
30 #[cfg(feature = "std")]
31 use lightning::events::{EventHandler, EventsProvider};
32 use lightning::ln::channelmanager::ChannelManager;
33 use lightning::ln::peer_handler::APeerManager;
34 use lightning::routing::gossip::{NetworkGraph, P2PGossipSync};
35 use lightning::routing::utxo::UtxoLookup;
36 use lightning::routing::router::Router;
37 use lightning::routing::scoring::{ScoreUpdate, WriteableScore};
38 use lightning::util::logger::Logger;
39 use lightning::util::persist::Persister;
40 #[cfg(feature = "std")]
41 use lightning::util::wakers::Sleeper;
42 use lightning_rapid_gossip_sync::RapidGossipSync;
45 use core::time::Duration;
47 #[cfg(feature = "std")]
49 #[cfg(feature = "std")]
50 use core::sync::atomic::{AtomicBool, Ordering};
51 #[cfg(feature = "std")]
52 use std::thread::{self, JoinHandle};
53 #[cfg(feature = "std")]
54 use std::time::Instant;
56 #[cfg(not(feature = "std"))]
59 /// `BackgroundProcessor` takes care of tasks that (1) need to happen periodically to keep
60 /// Rust-Lightning running properly, and (2) either can or should be run in the background. Its
61 /// responsibilities are:
62 /// * Processing [`Event`]s with a user-provided [`EventHandler`].
63 /// * Monitoring whether the [`ChannelManager`] needs to be re-persisted to disk, and if so,
64 /// writing it to disk/backups by invoking the callback given to it at startup.
65 /// [`ChannelManager`] persistence should be done in the background.
66 /// * Calling [`ChannelManager::timer_tick_occurred`], [`ChainMonitor::rebroadcast_pending_claims`]
67 /// and [`PeerManager::timer_tick_occurred`] at the appropriate intervals.
68 /// * Calling [`NetworkGraph::remove_stale_channels_and_tracking`] (if a [`GossipSync`] with a
69 /// [`NetworkGraph`] is provided to [`BackgroundProcessor::start`]).
71 /// It will also call [`PeerManager::process_events`] periodically though this shouldn't be relied
72 /// upon as doing so may result in high latency.
76 /// If [`ChannelManager`] persistence fails and the persisted manager becomes out-of-date, then
77 /// there is a risk of channels force-closing on startup when the manager realizes it's outdated.
78 /// However, as long as [`ChannelMonitor`] backups are sound, no funds besides those used for
79 /// unilateral chain closure fees are at risk.
81 /// [`ChannelMonitor`]: lightning::chain::channelmonitor::ChannelMonitor
82 /// [`Event`]: lightning::events::Event
83 /// [`PeerManager::timer_tick_occurred`]: lightning::ln::peer_handler::PeerManager::timer_tick_occurred
84 /// [`PeerManager::process_events`]: lightning::ln::peer_handler::PeerManager::process_events
85 #[cfg(feature = "std")]
86 #[must_use = "BackgroundProcessor will immediately stop on drop. It should be stored until shutdown."]
87 pub struct BackgroundProcessor {
88 stop_thread: Arc<AtomicBool>,
89 thread_handle: Option<JoinHandle<Result<(), std::io::Error>>>,
93 const FRESHNESS_TIMER: u64 = 60;
95 const FRESHNESS_TIMER: u64 = 1;
97 #[cfg(all(not(test), not(debug_assertions)))]
98 const PING_TIMER: u64 = 10;
99 /// Signature operations take a lot longer without compiler optimisations.
100 /// Increasing the ping timer allows for this but slower devices will be disconnected if the
101 /// timeout is reached.
102 #[cfg(all(not(test), debug_assertions))]
103 const PING_TIMER: u64 = 30;
105 const PING_TIMER: u64 = 1;
107 /// Prune the network graph of stale entries hourly.
108 const NETWORK_PRUNE_TIMER: u64 = 60 * 60;
111 const SCORER_PERSIST_TIMER: u64 = 60 * 60;
113 const SCORER_PERSIST_TIMER: u64 = 1;
116 const FIRST_NETWORK_PRUNE_TIMER: u64 = 60;
118 const FIRST_NETWORK_PRUNE_TIMER: u64 = 1;
121 const REBROADCAST_TIMER: u64 = 30;
123 const REBROADCAST_TIMER: u64 = 1;
125 #[cfg(feature = "futures")]
126 /// core::cmp::min is not currently const, so we define a trivial (and equivalent) replacement
127 const fn min_u64(a: u64, b: u64) -> u64 { if a < b { a } else { b } }
128 #[cfg(feature = "futures")]
129 const FASTEST_TIMER: u64 = min_u64(min_u64(FRESHNESS_TIMER, PING_TIMER),
130 min_u64(SCORER_PERSIST_TIMER, min_u64(FIRST_NETWORK_PRUNE_TIMER, REBROADCAST_TIMER)));
132 /// Either [`P2PGossipSync`] or [`RapidGossipSync`].
134 P: Deref<Target = P2PGossipSync<G, U, L>>,
135 R: Deref<Target = RapidGossipSync<G, L>>,
136 G: Deref<Target = NetworkGraph<L>>,
140 where U::Target: UtxoLookup, L::Target: Logger {
141 /// Gossip sync via the lightning peer-to-peer network as defined by BOLT 7.
143 /// Rapid gossip sync from a trusted server.
150 P: Deref<Target = P2PGossipSync<G, U, L>>,
151 R: Deref<Target = RapidGossipSync<G, L>>,
152 G: Deref<Target = NetworkGraph<L>>,
155 > GossipSync<P, R, G, U, L>
156 where U::Target: UtxoLookup, L::Target: Logger {
157 fn network_graph(&self) -> Option<&G> {
159 GossipSync::P2P(gossip_sync) => Some(gossip_sync.network_graph()),
160 GossipSync::Rapid(gossip_sync) => Some(gossip_sync.network_graph()),
161 GossipSync::None => None,
165 fn prunable_network_graph(&self) -> Option<&G> {
167 GossipSync::P2P(gossip_sync) => Some(gossip_sync.network_graph()),
168 GossipSync::Rapid(gossip_sync) => {
169 if gossip_sync.is_initial_sync_complete() {
170 Some(gossip_sync.network_graph())
175 GossipSync::None => None,
180 /// This is not exported to bindings users as the bindings concretize everything and have constructors for us
181 impl<P: Deref<Target = P2PGossipSync<G, U, L>>, G: Deref<Target = NetworkGraph<L>>, U: Deref, L: Deref>
182 GossipSync<P, &RapidGossipSync<G, L>, G, U, L>
184 U::Target: UtxoLookup,
187 /// Initializes a new [`GossipSync::P2P`] variant.
188 pub fn p2p(gossip_sync: P) -> Self {
189 GossipSync::P2P(gossip_sync)
193 /// This is not exported to bindings users as the bindings concretize everything and have constructors for us
194 impl<'a, R: Deref<Target = RapidGossipSync<G, L>>, G: Deref<Target = NetworkGraph<L>>, L: Deref>
196 &P2PGossipSync<G, &'a (dyn UtxoLookup + Send + Sync), L>,
199 &'a (dyn UtxoLookup + Send + Sync),
205 /// Initializes a new [`GossipSync::Rapid`] variant.
206 pub fn rapid(gossip_sync: R) -> Self {
207 GossipSync::Rapid(gossip_sync)
211 /// This is not exported to bindings users as the bindings concretize everything and have constructors for us
214 &P2PGossipSync<&'a NetworkGraph<L>, &'a (dyn UtxoLookup + Send + Sync), L>,
215 &RapidGossipSync<&'a NetworkGraph<L>, L>,
217 &'a (dyn UtxoLookup + Send + Sync),
223 /// Initializes a new [`GossipSync::None`] variant.
224 pub fn none() -> Self {
229 fn handle_network_graph_update<L: Deref>(
230 network_graph: &NetworkGraph<L>, event: &Event
231 ) where L::Target: Logger {
232 if let Event::PaymentPathFailed {
233 failure: PathFailure::OnPath { network_update: Some(ref upd) }, .. } = event
235 network_graph.handle_network_update(upd);
239 /// Updates scorer based on event and returns whether an update occurred so we can decide whether
241 fn update_scorer<'a, S: 'static + Deref<Target = SC> + Send + Sync, SC: 'a + WriteableScore<'a>>(
242 scorer: &'a S, event: &Event
245 Event::PaymentPathFailed { ref path, short_channel_id: Some(scid), .. } => {
246 let mut score = scorer.write_lock();
247 score.payment_path_failed(path, *scid);
249 Event::PaymentPathFailed { ref path, payment_failed_permanently: true, .. } => {
250 // Reached if the destination explicitly failed it back. We treat this as a successful probe
251 // because the payment made it all the way to the destination with sufficient liquidity.
252 let mut score = scorer.write_lock();
253 score.probe_successful(path);
255 Event::PaymentPathSuccessful { path, .. } => {
256 let mut score = scorer.write_lock();
257 score.payment_path_successful(path);
259 Event::ProbeSuccessful { path, .. } => {
260 let mut score = scorer.write_lock();
261 score.probe_successful(path);
263 Event::ProbeFailed { path, short_channel_id: Some(scid), .. } => {
264 let mut score = scorer.write_lock();
265 score.probe_failed(path, *scid);
272 macro_rules! define_run_body {
273 ($persister: ident, $chain_monitor: ident, $process_chain_monitor_events: expr,
274 $channel_manager: ident, $process_channel_manager_events: expr,
275 $gossip_sync: ident, $peer_manager: ident, $logger: ident, $scorer: ident,
276 $loop_exit_check: expr, $await: expr, $get_timer: expr, $timer_elapsed: expr,
277 $check_slow_await: expr)
279 log_trace!($logger, "Calling ChannelManager's timer_tick_occurred on startup");
280 $channel_manager.timer_tick_occurred();
281 log_trace!($logger, "Rebroadcasting monitor's pending claims on startup");
282 $chain_monitor.rebroadcast_pending_claims();
284 let mut last_freshness_call = $get_timer(FRESHNESS_TIMER);
285 let mut last_ping_call = $get_timer(PING_TIMER);
286 let mut last_prune_call = $get_timer(FIRST_NETWORK_PRUNE_TIMER);
287 let mut last_scorer_persist_call = $get_timer(SCORER_PERSIST_TIMER);
288 let mut last_rebroadcast_call = $get_timer(REBROADCAST_TIMER);
289 let mut have_pruned = false;
292 $process_channel_manager_events;
293 $process_chain_monitor_events;
295 // Note that the PeerManager::process_events may block on ChannelManager's locks,
296 // hence it comes last here. When the ChannelManager finishes whatever it's doing,
297 // we want to ensure we get into `persist_manager` as quickly as we can, especially
298 // without running the normal event processing above and handing events to users.
300 // Specifically, on an *extremely* slow machine, we may see ChannelManager start
301 // processing a message effectively at any point during this loop. In order to
302 // minimize the time between such processing completing and persisting the updated
303 // ChannelManager, we want to minimize methods blocking on a ChannelManager
304 // generally, and as a fallback place such blocking only immediately before
306 $peer_manager.as_ref().process_events();
308 // Exit the loop if the background processor was requested to stop.
309 if $loop_exit_check {
310 log_trace!($logger, "Terminating background processor.");
314 // We wait up to 100ms, but track how long it takes to detect being put to sleep,
315 // see `await_start`'s use below.
316 let mut await_start = None;
317 if $check_slow_await { await_start = Some($get_timer(1)); }
318 let updates_available = $await;
319 let await_slow = if $check_slow_await { $timer_elapsed(&mut await_start.unwrap(), 1) } else { false };
321 // Exit the loop if the background processor was requested to stop.
322 if $loop_exit_check {
323 log_trace!($logger, "Terminating background processor.");
327 if updates_available {
328 log_trace!($logger, "Persisting ChannelManager...");
329 $persister.persist_manager(&*$channel_manager)?;
330 log_trace!($logger, "Done persisting ChannelManager.");
332 if $timer_elapsed(&mut last_freshness_call, FRESHNESS_TIMER) {
333 log_trace!($logger, "Calling ChannelManager's timer_tick_occurred");
334 $channel_manager.timer_tick_occurred();
335 last_freshness_call = $get_timer(FRESHNESS_TIMER);
338 // On various platforms, we may be starved of CPU cycles for several reasons.
339 // E.g. on iOS, if we've been in the background, we will be entirely paused.
340 // Similarly, if we're on a desktop platform and the device has been asleep, we
341 // may not get any cycles.
342 // We detect this by checking if our max-100ms-sleep, above, ran longer than a
343 // full second, at which point we assume sockets may have been killed (they
344 // appear to be at least on some platforms, even if it has only been a second).
345 // Note that we have to take care to not get here just because user event
346 // processing was slow at the top of the loop. For example, the sample client
347 // may call Bitcoin Core RPCs during event handling, which very often takes
348 // more than a handful of seconds to complete, and shouldn't disconnect all our
350 log_trace!($logger, "100ms sleep took more than a second, disconnecting peers.");
351 $peer_manager.as_ref().disconnect_all_peers();
352 last_ping_call = $get_timer(PING_TIMER);
353 } else if $timer_elapsed(&mut last_ping_call, PING_TIMER) {
354 log_trace!($logger, "Calling PeerManager's timer_tick_occurred");
355 $peer_manager.as_ref().timer_tick_occurred();
356 last_ping_call = $get_timer(PING_TIMER);
359 // Note that we want to run a graph prune once not long after startup before
360 // falling back to our usual hourly prunes. This avoids short-lived clients never
361 // pruning their network graph. We run once 60 seconds after startup before
362 // continuing our normal cadence. For RGS, since 60 seconds is likely too long,
363 // we prune after an initial sync completes.
364 let prune_timer = if have_pruned { NETWORK_PRUNE_TIMER } else { FIRST_NETWORK_PRUNE_TIMER };
365 let prune_timer_elapsed = $timer_elapsed(&mut last_prune_call, prune_timer);
366 let should_prune = match $gossip_sync {
367 GossipSync::Rapid(_) => !have_pruned || prune_timer_elapsed,
368 _ => prune_timer_elapsed,
371 // The network graph must not be pruned while rapid sync completion is pending
372 if let Some(network_graph) = $gossip_sync.prunable_network_graph() {
373 #[cfg(feature = "std")] {
374 log_trace!($logger, "Pruning and persisting network graph.");
375 network_graph.remove_stale_channels_and_tracking();
377 #[cfg(not(feature = "std"))] {
378 log_warn!($logger, "Not pruning network graph, consider enabling `std` or doing so manually with remove_stale_channels_and_tracking_with_time.");
379 log_trace!($logger, "Persisting network graph.");
382 if let Err(e) = $persister.persist_graph(network_graph) {
383 log_error!($logger, "Error: Failed to persist network graph, check your disk and permissions {}", e)
388 let prune_timer = if have_pruned { NETWORK_PRUNE_TIMER } else { FIRST_NETWORK_PRUNE_TIMER };
389 last_prune_call = $get_timer(prune_timer);
392 if $timer_elapsed(&mut last_scorer_persist_call, SCORER_PERSIST_TIMER) {
393 if let Some(ref scorer) = $scorer {
394 log_trace!($logger, "Persisting scorer");
395 if let Err(e) = $persister.persist_scorer(&scorer) {
396 log_error!($logger, "Error: Failed to persist scorer, check your disk and permissions {}", e)
399 last_scorer_persist_call = $get_timer(SCORER_PERSIST_TIMER);
402 if $timer_elapsed(&mut last_rebroadcast_call, REBROADCAST_TIMER) {
403 log_trace!($logger, "Rebroadcasting monitor's pending claims");
404 $chain_monitor.rebroadcast_pending_claims();
405 last_rebroadcast_call = $get_timer(REBROADCAST_TIMER);
409 // After we exit, ensure we persist the ChannelManager one final time - this avoids
410 // some races where users quit while channel updates were in-flight, with
411 // ChannelMonitor update(s) persisted without a corresponding ChannelManager update.
412 $persister.persist_manager(&*$channel_manager)?;
414 // Persist Scorer on exit
415 if let Some(ref scorer) = $scorer {
416 $persister.persist_scorer(&scorer)?;
419 // Persist NetworkGraph on exit
420 if let Some(network_graph) = $gossip_sync.network_graph() {
421 $persister.persist_graph(network_graph)?;
428 #[cfg(feature = "futures")]
429 pub(crate) mod futures_util {
430 use core::future::Future;
431 use core::task::{Poll, Waker, RawWaker, RawWakerVTable};
433 use core::marker::Unpin;
434 pub(crate) struct Selector<
435 A: Future<Output=()> + Unpin, B: Future<Output=()> + Unpin, C: Future<Output=bool> + Unpin
441 pub(crate) enum SelectorOutput {
446 A: Future<Output=()> + Unpin, B: Future<Output=()> + Unpin, C: Future<Output=bool> + Unpin
447 > Future for Selector<A, B, C> {
448 type Output = SelectorOutput;
449 fn poll(mut self: Pin<&mut Self>, ctx: &mut core::task::Context<'_>) -> Poll<SelectorOutput> {
450 match Pin::new(&mut self.a).poll(ctx) {
451 Poll::Ready(()) => { return Poll::Ready(SelectorOutput::A); },
454 match Pin::new(&mut self.b).poll(ctx) {
455 Poll::Ready(()) => { return Poll::Ready(SelectorOutput::B); },
458 match Pin::new(&mut self.c).poll(ctx) {
459 Poll::Ready(res) => { return Poll::Ready(SelectorOutput::C(res)); },
466 // If we want to poll a future without an async context to figure out if it has completed or
467 // not without awaiting, we need a Waker, which needs a vtable...we fill it with dummy values
468 // but sadly there's a good bit of boilerplate here.
469 fn dummy_waker_clone(_: *const ()) -> RawWaker { RawWaker::new(core::ptr::null(), &DUMMY_WAKER_VTABLE) }
470 fn dummy_waker_action(_: *const ()) { }
472 const DUMMY_WAKER_VTABLE: RawWakerVTable = RawWakerVTable::new(
473 dummy_waker_clone, dummy_waker_action, dummy_waker_action, dummy_waker_action);
474 pub(crate) fn dummy_waker() -> Waker { unsafe { Waker::from_raw(RawWaker::new(core::ptr::null(), &DUMMY_WAKER_VTABLE)) } }
476 #[cfg(feature = "futures")]
477 use futures_util::{Selector, SelectorOutput, dummy_waker};
478 #[cfg(feature = "futures")]
481 /// Processes background events in a future.
483 /// `sleeper` should return a future which completes in the given amount of time and returns a
484 /// boolean indicating whether the background processing should exit. Once `sleeper` returns a
485 /// future which outputs `true`, the loop will exit and this function's future will complete.
486 /// The `sleeper` future is free to return early after it has triggered the exit condition.
488 /// See [`BackgroundProcessor::start`] for information on which actions this handles.
490 /// Requires the `futures` feature. Note that while this method is available without the `std`
491 /// feature, doing so will skip calling [`NetworkGraph::remove_stale_channels_and_tracking`],
492 /// you should call [`NetworkGraph::remove_stale_channels_and_tracking_with_time`] regularly
493 /// manually instead.
495 /// The `mobile_interruptable_platform` flag should be set if we're currently running on a
496 /// mobile device, where we may need to check for interruption of the application regularly. If you
497 /// are unsure, you should set the flag, as the performance impact of it is minimal unless there
498 /// are hundreds or thousands of simultaneous process calls running.
500 /// For example, in order to process background events in a [Tokio](https://tokio.rs/) task, you
501 /// could setup `process_events_async` like this:
503 /// # struct MyPersister {}
504 /// # impl lightning::util::persist::KVStorePersister for MyPersister {
505 /// # fn persist<W: lightning::util::ser::Writeable>(&self, key: &str, object: &W) -> lightning::io::Result<()> { Ok(()) }
507 /// # struct MyEventHandler {}
508 /// # impl MyEventHandler {
509 /// # async fn handle_event(&self, _: lightning::events::Event) {}
511 /// # #[derive(Eq, PartialEq, Clone, Hash)]
512 /// # struct MySocketDescriptor {}
513 /// # impl lightning::ln::peer_handler::SocketDescriptor for MySocketDescriptor {
514 /// # fn send_data(&mut self, _data: &[u8], _resume_read: bool) -> usize { 0 }
515 /// # fn disconnect_socket(&mut self) {}
517 /// # use std::sync::{Arc, Mutex};
518 /// # use std::sync::atomic::{AtomicBool, Ordering};
519 /// # use lightning_background_processor::{process_events_async, GossipSync};
520 /// # type MyBroadcaster = dyn lightning::chain::chaininterface::BroadcasterInterface + Send + Sync;
521 /// # type MyFeeEstimator = dyn lightning::chain::chaininterface::FeeEstimator + Send + Sync;
522 /// # type MyNodeSigner = dyn lightning::sign::NodeSigner + Send + Sync;
523 /// # type MyUtxoLookup = dyn lightning::routing::utxo::UtxoLookup + Send + Sync;
524 /// # type MyFilter = dyn lightning::chain::Filter + Send + Sync;
525 /// # type MyLogger = dyn lightning::util::logger::Logger + Send + Sync;
526 /// # type MyChainMonitor = lightning::chain::chainmonitor::ChainMonitor<lightning::sign::InMemorySigner, Arc<MyFilter>, Arc<MyBroadcaster>, Arc<MyFeeEstimator>, Arc<MyLogger>, Arc<MyPersister>>;
527 /// # type MyPeerManager = lightning::ln::peer_handler::SimpleArcPeerManager<MySocketDescriptor, MyChainMonitor, MyBroadcaster, MyFeeEstimator, MyUtxoLookup, MyLogger>;
528 /// # type MyNetworkGraph = lightning::routing::gossip::NetworkGraph<Arc<MyLogger>>;
529 /// # type MyGossipSync = lightning::routing::gossip::P2PGossipSync<Arc<MyNetworkGraph>, Arc<MyUtxoLookup>, Arc<MyLogger>>;
530 /// # type MyChannelManager = lightning::ln::channelmanager::SimpleArcChannelManager<MyChainMonitor, MyBroadcaster, MyFeeEstimator, MyLogger>;
531 /// # type MyScorer = Mutex<lightning::routing::scoring::ProbabilisticScorer<Arc<MyNetworkGraph>, Arc<MyLogger>>>;
533 /// # async fn setup_background_processing(my_persister: Arc<MyPersister>, my_event_handler: Arc<MyEventHandler>, my_chain_monitor: Arc<MyChainMonitor>, my_channel_manager: Arc<MyChannelManager>, my_gossip_sync: Arc<MyGossipSync>, my_logger: Arc<MyLogger>, my_scorer: Arc<MyScorer>, my_peer_manager: Arc<MyPeerManager>) {
534 /// let background_persister = Arc::clone(&my_persister);
535 /// let background_event_handler = Arc::clone(&my_event_handler);
536 /// let background_chain_mon = Arc::clone(&my_chain_monitor);
537 /// let background_chan_man = Arc::clone(&my_channel_manager);
538 /// let background_gossip_sync = GossipSync::p2p(Arc::clone(&my_gossip_sync));
539 /// let background_peer_man = Arc::clone(&my_peer_manager);
540 /// let background_logger = Arc::clone(&my_logger);
541 /// let background_scorer = Arc::clone(&my_scorer);
543 /// // Setup the sleeper.
544 /// let (stop_sender, stop_receiver) = tokio::sync::watch::channel(());
546 /// let sleeper = move |d| {
547 /// let mut receiver = stop_receiver.clone();
548 /// Box::pin(async move {
550 /// _ = tokio::time::sleep(d) => false,
551 /// _ = receiver.changed() => true,
556 /// let mobile_interruptable_platform = false;
558 /// let handle = tokio::spawn(async move {
559 /// process_events_async(
560 /// background_persister,
561 /// |e| background_event_handler.handle_event(e),
562 /// background_chain_mon,
563 /// background_chan_man,
564 /// background_gossip_sync,
565 /// background_peer_man,
566 /// background_logger,
567 /// Some(background_scorer),
569 /// mobile_interruptable_platform,
572 /// .expect("Failed to process events");
575 /// // Stop the background processing.
576 /// stop_sender.send(()).unwrap();
577 /// handle.await.unwrap();
580 #[cfg(feature = "futures")]
581 pub async fn process_events_async<
583 UL: 'static + Deref + Send + Sync,
584 CF: 'static + Deref + Send + Sync,
585 CW: 'static + Deref + Send + Sync,
586 T: 'static + Deref + Send + Sync,
587 ES: 'static + Deref + Send + Sync,
588 NS: 'static + Deref + Send + Sync,
589 SP: 'static + Deref + Send + Sync,
590 F: 'static + Deref + Send + Sync,
591 R: 'static + Deref + Send + Sync,
592 G: 'static + Deref<Target = NetworkGraph<L>> + Send + Sync,
593 L: 'static + Deref + Send + Sync,
594 P: 'static + Deref + Send + Sync,
595 EventHandlerFuture: core::future::Future<Output = ()>,
596 EventHandler: Fn(Event) -> EventHandlerFuture,
597 PS: 'static + Deref + Send,
598 M: 'static + Deref<Target = ChainMonitor<<SP::Target as SignerProvider>::Signer, CF, T, F, L, P>> + Send + Sync,
599 CM: 'static + Deref<Target = ChannelManager<CW, T, ES, NS, SP, F, R, L>> + Send + Sync,
600 PGS: 'static + Deref<Target = P2PGossipSync<G, UL, L>> + Send + Sync,
601 RGS: 'static + Deref<Target = RapidGossipSync<G, L>> + Send,
602 APM: APeerManager + Send + Sync,
603 PM: 'static + Deref<Target = APM> + Send + Sync,
604 S: 'static + Deref<Target = SC> + Send + Sync,
605 SC: for<'b> WriteableScore<'b>,
606 SleepFuture: core::future::Future<Output = bool> + core::marker::Unpin,
607 Sleeper: Fn(Duration) -> SleepFuture
609 persister: PS, event_handler: EventHandler, chain_monitor: M, channel_manager: CM,
610 gossip_sync: GossipSync<PGS, RGS, G, UL, L>, peer_manager: PM, logger: L, scorer: Option<S>,
611 sleeper: Sleeper, mobile_interruptable_platform: bool,
612 ) -> Result<(), lightning::io::Error>
614 UL::Target: 'static + UtxoLookup,
615 CF::Target: 'static + chain::Filter,
616 CW::Target: 'static + chain::Watch<<SP::Target as SignerProvider>::Signer>,
617 T::Target: 'static + BroadcasterInterface,
618 ES::Target: 'static + EntropySource,
619 NS::Target: 'static + NodeSigner,
620 SP::Target: 'static + SignerProvider,
621 F::Target: 'static + FeeEstimator,
622 R::Target: 'static + Router,
623 L::Target: 'static + Logger,
624 P::Target: 'static + Persist<<SP::Target as SignerProvider>::Signer>,
625 PS::Target: 'static + Persister<'a, CW, T, ES, NS, SP, F, R, L, SC>,
627 let mut should_break = false;
628 let async_event_handler = |event| {
629 let network_graph = gossip_sync.network_graph();
630 let event_handler = &event_handler;
631 let scorer = &scorer;
632 let logger = &logger;
633 let persister = &persister;
635 if let Some(network_graph) = network_graph {
636 handle_network_graph_update(network_graph, &event)
638 if let Some(ref scorer) = scorer {
639 if update_scorer(scorer, &event) {
640 log_trace!(logger, "Persisting scorer after update");
641 if let Err(e) = persister.persist_scorer(&scorer) {
642 log_error!(logger, "Error: Failed to persist scorer, check your disk and permissions {}", e)
646 event_handler(event).await;
649 define_run_body!(persister,
650 chain_monitor, chain_monitor.process_pending_events_async(async_event_handler).await,
651 channel_manager, channel_manager.process_pending_events_async(async_event_handler).await,
652 gossip_sync, peer_manager, logger, scorer, should_break, {
654 a: channel_manager.get_persistable_update_future(),
655 b: chain_monitor.get_update_future(),
656 c: sleeper(if mobile_interruptable_platform { Duration::from_millis(100) } else { Duration::from_secs(FASTEST_TIMER) }),
659 SelectorOutput::A => true,
660 SelectorOutput::B => false,
661 SelectorOutput::C(exit) => {
666 }, |t| sleeper(Duration::from_secs(t)),
667 |fut: &mut SleepFuture, _| {
668 let mut waker = dummy_waker();
669 let mut ctx = task::Context::from_waker(&mut waker);
670 match core::pin::Pin::new(fut).poll(&mut ctx) {
671 task::Poll::Ready(exit) => { should_break = exit; true },
672 task::Poll::Pending => false,
674 }, mobile_interruptable_platform)
677 #[cfg(feature = "std")]
678 impl BackgroundProcessor {
679 /// Start a background thread that takes care of responsibilities enumerated in the [top-level
682 /// The thread runs indefinitely unless the object is dropped, [`stop`] is called, or
683 /// [`Persister::persist_manager`] returns an error. In case of an error, the error is retrieved by calling
684 /// either [`join`] or [`stop`].
686 /// # Data Persistence
688 /// [`Persister::persist_manager`] is responsible for writing out the [`ChannelManager`] to disk, and/or
689 /// uploading to one or more backup services. See [`ChannelManager::write`] for writing out a
690 /// [`ChannelManager`]. See the `lightning-persister` crate for LDK's
691 /// provided implementation.
693 /// [`Persister::persist_graph`] is responsible for writing out the [`NetworkGraph`] to disk, if
694 /// [`GossipSync`] is supplied. See [`NetworkGraph::write`] for writing out a [`NetworkGraph`].
695 /// See the `lightning-persister` crate for LDK's provided implementation.
697 /// Typically, users should either implement [`Persister::persist_manager`] to never return an
698 /// error or call [`join`] and handle any error that may arise. For the latter case,
699 /// `BackgroundProcessor` must be restarted by calling `start` again after handling the error.
703 /// `event_handler` is responsible for handling events that users should be notified of (e.g.,
704 /// payment failed). [`BackgroundProcessor`] may decorate the given [`EventHandler`] with common
705 /// functionality implemented by other handlers.
706 /// * [`P2PGossipSync`] if given will update the [`NetworkGraph`] based on payment failures.
708 /// # Rapid Gossip Sync
710 /// If rapid gossip sync is meant to run at startup, pass [`RapidGossipSync`] via `gossip_sync`
711 /// to indicate that the [`BackgroundProcessor`] should not prune the [`NetworkGraph`] instance
712 /// until the [`RapidGossipSync`] instance completes its first sync.
714 /// [top-level documentation]: BackgroundProcessor
715 /// [`join`]: Self::join
716 /// [`stop`]: Self::stop
717 /// [`ChannelManager`]: lightning::ln::channelmanager::ChannelManager
718 /// [`ChannelManager::write`]: lightning::ln::channelmanager::ChannelManager#impl-Writeable
719 /// [`Persister::persist_manager`]: lightning::util::persist::Persister::persist_manager
720 /// [`Persister::persist_graph`]: lightning::util::persist::Persister::persist_graph
721 /// [`NetworkGraph`]: lightning::routing::gossip::NetworkGraph
722 /// [`NetworkGraph::write`]: lightning::routing::gossip::NetworkGraph#impl-Writeable
725 UL: 'static + Deref + Send + Sync,
726 CF: 'static + Deref + Send + Sync,
727 CW: 'static + Deref + Send + Sync,
728 T: 'static + Deref + Send + Sync,
729 ES: 'static + Deref + Send + Sync,
730 NS: 'static + Deref + Send + Sync,
731 SP: 'static + Deref + Send + Sync,
732 F: 'static + Deref + Send + Sync,
733 R: 'static + Deref + Send + Sync,
734 G: 'static + Deref<Target = NetworkGraph<L>> + Send + Sync,
735 L: 'static + Deref + Send + Sync,
736 P: 'static + Deref + Send + Sync,
737 EH: 'static + EventHandler + Send,
738 PS: 'static + Deref + Send,
739 M: 'static + Deref<Target = ChainMonitor<<SP::Target as SignerProvider>::Signer, CF, T, F, L, P>> + Send + Sync,
740 CM: 'static + Deref<Target = ChannelManager<CW, T, ES, NS, SP, F, R, L>> + Send + Sync,
741 PGS: 'static + Deref<Target = P2PGossipSync<G, UL, L>> + Send + Sync,
742 RGS: 'static + Deref<Target = RapidGossipSync<G, L>> + Send,
743 APM: APeerManager + Send + Sync,
744 PM: 'static + Deref<Target = APM> + Send + Sync,
745 S: 'static + Deref<Target = SC> + Send + Sync,
746 SC: for <'b> WriteableScore<'b>,
748 persister: PS, event_handler: EH, chain_monitor: M, channel_manager: CM,
749 gossip_sync: GossipSync<PGS, RGS, G, UL, L>, peer_manager: PM, logger: L, scorer: Option<S>,
752 UL::Target: 'static + UtxoLookup,
753 CF::Target: 'static + chain::Filter,
754 CW::Target: 'static + chain::Watch<<SP::Target as SignerProvider>::Signer>,
755 T::Target: 'static + BroadcasterInterface,
756 ES::Target: 'static + EntropySource,
757 NS::Target: 'static + NodeSigner,
758 SP::Target: 'static + SignerProvider,
759 F::Target: 'static + FeeEstimator,
760 R::Target: 'static + Router,
761 L::Target: 'static + Logger,
762 P::Target: 'static + Persist<<SP::Target as SignerProvider>::Signer>,
763 PS::Target: 'static + Persister<'a, CW, T, ES, NS, SP, F, R, L, SC>,
765 let stop_thread = Arc::new(AtomicBool::new(false));
766 let stop_thread_clone = stop_thread.clone();
767 let handle = thread::spawn(move || -> Result<(), std::io::Error> {
768 let event_handler = |event| {
769 let network_graph = gossip_sync.network_graph();
770 if let Some(network_graph) = network_graph {
771 handle_network_graph_update(network_graph, &event)
773 if let Some(ref scorer) = scorer {
774 if update_scorer(scorer, &event) {
775 log_trace!(logger, "Persisting scorer after update");
776 if let Err(e) = persister.persist_scorer(&scorer) {
777 log_error!(logger, "Error: Failed to persist scorer, check your disk and permissions {}", e)
781 event_handler.handle_event(event);
783 define_run_body!(persister, chain_monitor, chain_monitor.process_pending_events(&event_handler),
784 channel_manager, channel_manager.process_pending_events(&event_handler),
785 gossip_sync, peer_manager, logger, scorer, stop_thread.load(Ordering::Acquire),
786 Sleeper::from_two_futures(
787 channel_manager.get_persistable_update_future(),
788 chain_monitor.get_update_future()
789 ).wait_timeout(Duration::from_millis(100)),
790 |_| Instant::now(), |time: &Instant, dur| time.elapsed().as_secs() > dur, false)
792 Self { stop_thread: stop_thread_clone, thread_handle: Some(handle) }
795 /// Join `BackgroundProcessor`'s thread, returning any error that occurred while persisting
796 /// [`ChannelManager`].
800 /// This function panics if the background thread has panicked such as while persisting or
803 /// [`ChannelManager`]: lightning::ln::channelmanager::ChannelManager
804 pub fn join(mut self) -> Result<(), std::io::Error> {
805 assert!(self.thread_handle.is_some());
809 /// Stop `BackgroundProcessor`'s thread, returning any error that occurred while persisting
810 /// [`ChannelManager`].
814 /// This function panics if the background thread has panicked such as while persisting or
817 /// [`ChannelManager`]: lightning::ln::channelmanager::ChannelManager
818 pub fn stop(mut self) -> Result<(), std::io::Error> {
819 assert!(self.thread_handle.is_some());
820 self.stop_and_join_thread()
823 fn stop_and_join_thread(&mut self) -> Result<(), std::io::Error> {
824 self.stop_thread.store(true, Ordering::Release);
828 fn join_thread(&mut self) -> Result<(), std::io::Error> {
829 match self.thread_handle.take() {
830 Some(handle) => handle.join().unwrap(),
836 #[cfg(feature = "std")]
837 impl Drop for BackgroundProcessor {
839 self.stop_and_join_thread().unwrap();
843 #[cfg(all(feature = "std", test))]
845 use bitcoin::blockdata::constants::{genesis_block, ChainHash};
846 use bitcoin::blockdata::locktime::PackedLockTime;
847 use bitcoin::blockdata::transaction::{Transaction, TxOut};
848 use bitcoin::network::constants::Network;
849 use bitcoin::secp256k1::{SecretKey, PublicKey, Secp256k1};
850 use lightning::chain::{BestBlock, Confirm, chainmonitor};
851 use lightning::chain::channelmonitor::ANTI_REORG_DELAY;
852 use lightning::sign::{InMemorySigner, KeysManager};
853 use lightning::chain::transaction::OutPoint;
854 use lightning::events::{Event, PathFailure, MessageSendEventsProvider, MessageSendEvent};
855 use lightning::{get_event_msg, get_event};
856 use lightning::ln::PaymentHash;
857 use lightning::ln::channelmanager;
858 use lightning::ln::channelmanager::{BREAKDOWN_TIMEOUT, ChainParameters, MIN_CLTV_EXPIRY_DELTA, PaymentId};
859 use lightning::ln::features::{ChannelFeatures, NodeFeatures};
860 use lightning::ln::functional_test_utils::*;
861 use lightning::ln::msgs::{ChannelMessageHandler, Init};
862 use lightning::ln::peer_handler::{PeerManager, MessageHandler, SocketDescriptor, IgnoringMessageHandler};
863 use lightning::routing::gossip::{NetworkGraph, NodeId, P2PGossipSync};
864 use lightning::routing::router::{DefaultRouter, Path, RouteHop};
865 use lightning::routing::scoring::{ChannelUsage, ScoreUpdate, ScoreLookUp};
866 use lightning::util::config::UserConfig;
867 use lightning::util::ser::Writeable;
868 use lightning::util::test_utils;
869 use lightning::util::persist::KVStorePersister;
870 use lightning_persister::FilesystemPersister;
871 use std::collections::VecDeque;
873 use std::path::PathBuf;
874 use std::sync::{Arc, Mutex};
875 use std::sync::mpsc::SyncSender;
876 use std::time::Duration;
877 use lightning_rapid_gossip_sync::RapidGossipSync;
878 use super::{BackgroundProcessor, GossipSync, FRESHNESS_TIMER};
880 const EVENT_DEADLINE: u64 = 5 * FRESHNESS_TIMER;
882 #[derive(Clone, Hash, PartialEq, Eq)]
883 struct TestDescriptor{}
884 impl SocketDescriptor for TestDescriptor {
885 fn send_data(&mut self, _data: &[u8], _resume_read: bool) -> usize {
889 fn disconnect_socket(&mut self) {}
892 type ChannelManager =
893 channelmanager::ChannelManager<
895 Arc<test_utils::TestBroadcaster>,
899 Arc<test_utils::TestFeeEstimator>,
901 Arc<NetworkGraph<Arc<test_utils::TestLogger>>>,
902 Arc<test_utils::TestLogger>,
903 Arc<Mutex<TestScorer>>,
907 Arc<test_utils::TestLogger>>;
909 type ChainMonitor = chainmonitor::ChainMonitor<InMemorySigner, Arc<test_utils::TestChainSource>, Arc<test_utils::TestBroadcaster>, Arc<test_utils::TestFeeEstimator>, Arc<test_utils::TestLogger>, Arc<FilesystemPersister>>;
911 type PGS = Arc<P2PGossipSync<Arc<NetworkGraph<Arc<test_utils::TestLogger>>>, Arc<test_utils::TestChainSource>, Arc<test_utils::TestLogger>>>;
912 type RGS = Arc<RapidGossipSync<Arc<NetworkGraph<Arc<test_utils::TestLogger>>>, Arc<test_utils::TestLogger>>>;
915 node: Arc<ChannelManager>,
916 p2p_gossip_sync: PGS,
917 rapid_gossip_sync: RGS,
918 peer_manager: Arc<PeerManager<TestDescriptor, Arc<test_utils::TestChannelMessageHandler>, Arc<test_utils::TestRoutingMessageHandler>, IgnoringMessageHandler, Arc<test_utils::TestLogger>, IgnoringMessageHandler, Arc<KeysManager>>>,
919 chain_monitor: Arc<ChainMonitor>,
920 persister: Arc<FilesystemPersister>,
921 tx_broadcaster: Arc<test_utils::TestBroadcaster>,
922 network_graph: Arc<NetworkGraph<Arc<test_utils::TestLogger>>>,
923 logger: Arc<test_utils::TestLogger>,
924 best_block: BestBlock,
925 scorer: Arc<Mutex<TestScorer>>,
929 fn p2p_gossip_sync(&self) -> GossipSync<PGS, RGS, Arc<NetworkGraph<Arc<test_utils::TestLogger>>>, Arc<test_utils::TestChainSource>, Arc<test_utils::TestLogger>> {
930 GossipSync::P2P(self.p2p_gossip_sync.clone())
933 fn rapid_gossip_sync(&self) -> GossipSync<PGS, RGS, Arc<NetworkGraph<Arc<test_utils::TestLogger>>>, Arc<test_utils::TestChainSource>, Arc<test_utils::TestLogger>> {
934 GossipSync::Rapid(self.rapid_gossip_sync.clone())
937 fn no_gossip_sync(&self) -> GossipSync<PGS, RGS, Arc<NetworkGraph<Arc<test_utils::TestLogger>>>, Arc<test_utils::TestChainSource>, Arc<test_utils::TestLogger>> {
944 let data_dir = self.persister.get_data_dir();
945 match fs::remove_dir_all(data_dir.clone()) {
946 Err(e) => println!("Failed to remove test persister directory {}: {}", data_dir, e),
953 graph_error: Option<(std::io::ErrorKind, &'static str)>,
954 graph_persistence_notifier: Option<SyncSender<()>>,
955 manager_error: Option<(std::io::ErrorKind, &'static str)>,
956 scorer_error: Option<(std::io::ErrorKind, &'static str)>,
957 filesystem_persister: FilesystemPersister,
961 fn new(data_dir: String) -> Self {
962 let filesystem_persister = FilesystemPersister::new(data_dir);
963 Self { graph_error: None, graph_persistence_notifier: None, manager_error: None, scorer_error: None, filesystem_persister }
966 fn with_graph_error(self, error: std::io::ErrorKind, message: &'static str) -> Self {
967 Self { graph_error: Some((error, message)), ..self }
970 fn with_graph_persistence_notifier(self, sender: SyncSender<()>) -> Self {
971 Self { graph_persistence_notifier: Some(sender), ..self }
974 fn with_manager_error(self, error: std::io::ErrorKind, message: &'static str) -> Self {
975 Self { manager_error: Some((error, message)), ..self }
978 fn with_scorer_error(self, error: std::io::ErrorKind, message: &'static str) -> Self {
979 Self { scorer_error: Some((error, message)), ..self }
983 impl KVStorePersister for Persister {
984 fn persist<W: Writeable>(&self, key: &str, object: &W) -> std::io::Result<()> {
985 if key == "manager" {
986 if let Some((error, message)) = self.manager_error {
987 return Err(std::io::Error::new(error, message))
991 if key == "network_graph" {
992 if let Some(sender) = &self.graph_persistence_notifier {
993 match sender.send(()) {
995 Err(std::sync::mpsc::SendError(())) => println!("Persister failed to notify as receiver went away."),
999 if let Some((error, message)) = self.graph_error {
1000 return Err(std::io::Error::new(error, message))
1004 if key == "scorer" {
1005 if let Some((error, message)) = self.scorer_error {
1006 return Err(std::io::Error::new(error, message))
1010 self.filesystem_persister.persist(key, object)
1015 event_expectations: Option<VecDeque<TestResult>>,
1020 PaymentFailure { path: Path, short_channel_id: u64 },
1021 PaymentSuccess { path: Path },
1022 ProbeFailure { path: Path },
1023 ProbeSuccess { path: Path },
1028 Self { event_expectations: None }
1031 fn expect(&mut self, expectation: TestResult) {
1032 self.event_expectations.get_or_insert_with(VecDeque::new).push_back(expectation);
1036 impl lightning::util::ser::Writeable for TestScorer {
1037 fn write<W: lightning::util::ser::Writer>(&self, _: &mut W) -> Result<(), lightning::io::Error> { Ok(()) }
1040 impl ScoreLookUp for TestScorer {
1041 type ScoreParams = ();
1042 fn channel_penalty_msat(
1043 &self, _short_channel_id: u64, _source: &NodeId, _target: &NodeId, _usage: ChannelUsage, _score_params: &Self::ScoreParams
1044 ) -> u64 { unimplemented!(); }
1047 impl ScoreUpdate for TestScorer {
1048 fn payment_path_failed(&mut self, actual_path: &Path, actual_short_channel_id: u64) {
1049 if let Some(expectations) = &mut self.event_expectations {
1050 match expectations.pop_front().unwrap() {
1051 TestResult::PaymentFailure { path, short_channel_id } => {
1052 assert_eq!(actual_path, &path);
1053 assert_eq!(actual_short_channel_id, short_channel_id);
1055 TestResult::PaymentSuccess { path } => {
1056 panic!("Unexpected successful payment path: {:?}", path)
1058 TestResult::ProbeFailure { path } => {
1059 panic!("Unexpected probe failure: {:?}", path)
1061 TestResult::ProbeSuccess { path } => {
1062 panic!("Unexpected probe success: {:?}", path)
1068 fn payment_path_successful(&mut self, actual_path: &Path) {
1069 if let Some(expectations) = &mut self.event_expectations {
1070 match expectations.pop_front().unwrap() {
1071 TestResult::PaymentFailure { path, .. } => {
1072 panic!("Unexpected payment path failure: {:?}", path)
1074 TestResult::PaymentSuccess { path } => {
1075 assert_eq!(actual_path, &path);
1077 TestResult::ProbeFailure { path } => {
1078 panic!("Unexpected probe failure: {:?}", path)
1080 TestResult::ProbeSuccess { path } => {
1081 panic!("Unexpected probe success: {:?}", path)
1087 fn probe_failed(&mut self, actual_path: &Path, _: u64) {
1088 if let Some(expectations) = &mut self.event_expectations {
1089 match expectations.pop_front().unwrap() {
1090 TestResult::PaymentFailure { path, .. } => {
1091 panic!("Unexpected payment path failure: {:?}", path)
1093 TestResult::PaymentSuccess { path } => {
1094 panic!("Unexpected payment path success: {:?}", path)
1096 TestResult::ProbeFailure { path } => {
1097 assert_eq!(actual_path, &path);
1099 TestResult::ProbeSuccess { path } => {
1100 panic!("Unexpected probe success: {:?}", path)
1105 fn probe_successful(&mut self, actual_path: &Path) {
1106 if let Some(expectations) = &mut self.event_expectations {
1107 match expectations.pop_front().unwrap() {
1108 TestResult::PaymentFailure { path, .. } => {
1109 panic!("Unexpected payment path failure: {:?}", path)
1111 TestResult::PaymentSuccess { path } => {
1112 panic!("Unexpected payment path success: {:?}", path)
1114 TestResult::ProbeFailure { path } => {
1115 panic!("Unexpected probe failure: {:?}", path)
1117 TestResult::ProbeSuccess { path } => {
1118 assert_eq!(actual_path, &path);
1125 impl Drop for TestScorer {
1126 fn drop(&mut self) {
1127 if std::thread::panicking() {
1131 if let Some(event_expectations) = &self.event_expectations {
1132 if !event_expectations.is_empty() {
1133 panic!("Unsatisfied event expectations: {:?}", event_expectations);
1139 fn get_full_filepath(filepath: String, filename: String) -> String {
1140 let mut path = PathBuf::from(filepath);
1141 path.push(filename);
1142 path.to_str().unwrap().to_string()
1145 fn create_nodes(num_nodes: usize, persist_dir: &str) -> (String, Vec<Node>) {
1146 let persist_temp_path = env::temp_dir().join(persist_dir);
1147 let persist_dir = persist_temp_path.to_string_lossy().to_string();
1148 let network = Network::Bitcoin;
1149 let mut nodes = Vec::new();
1150 for i in 0..num_nodes {
1151 let tx_broadcaster = Arc::new(test_utils::TestBroadcaster::new(network));
1152 let fee_estimator = Arc::new(test_utils::TestFeeEstimator { sat_per_kw: Mutex::new(253) });
1153 let logger = Arc::new(test_utils::TestLogger::with_id(format!("node {}", i)));
1154 let genesis_block = genesis_block(network);
1155 let network_graph = Arc::new(NetworkGraph::new(network, logger.clone()));
1156 let scorer = Arc::new(Mutex::new(TestScorer::new()));
1157 let seed = [i as u8; 32];
1158 let router = Arc::new(DefaultRouter::new(network_graph.clone(), logger.clone(), seed, scorer.clone(), ()));
1159 let chain_source = Arc::new(test_utils::TestChainSource::new(Network::Bitcoin));
1160 let persister = Arc::new(FilesystemPersister::new(format!("{}_persister_{}", &persist_dir, i)));
1161 let now = Duration::from_secs(genesis_block.header.time as u64);
1162 let keys_manager = Arc::new(KeysManager::new(&seed, now.as_secs(), now.subsec_nanos()));
1163 let chain_monitor = Arc::new(chainmonitor::ChainMonitor::new(Some(chain_source.clone()), tx_broadcaster.clone(), logger.clone(), fee_estimator.clone(), persister.clone()));
1164 let best_block = BestBlock::from_network(network);
1165 let params = ChainParameters { network, best_block };
1166 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));
1167 let p2p_gossip_sync = Arc::new(P2PGossipSync::new(network_graph.clone(), Some(chain_source.clone()), logger.clone()));
1168 let rapid_gossip_sync = Arc::new(RapidGossipSync::new(network_graph.clone(), logger.clone()));
1169 let msg_handler = MessageHandler {
1170 chan_handler: Arc::new(test_utils::TestChannelMessageHandler::new(ChainHash::using_genesis_block(Network::Testnet))),
1171 route_handler: Arc::new(test_utils::TestRoutingMessageHandler::new()),
1172 onion_message_handler: IgnoringMessageHandler{}, custom_message_handler: IgnoringMessageHandler{}
1174 let peer_manager = Arc::new(PeerManager::new(msg_handler, 0, &seed, logger.clone(), keys_manager.clone()));
1175 let node = Node { node: manager, p2p_gossip_sync, rapid_gossip_sync, peer_manager, chain_monitor, persister, tx_broadcaster, network_graph, logger, best_block, scorer };
1179 for i in 0..num_nodes {
1180 for j in (i+1)..num_nodes {
1181 nodes[i].node.peer_connected(&nodes[j].node.get_our_node_id(), &Init {
1182 features: nodes[j].node.init_features(), networks: None, remote_network_address: None
1184 nodes[j].node.peer_connected(&nodes[i].node.get_our_node_id(), &Init {
1185 features: nodes[i].node.init_features(), networks: None, remote_network_address: None
1190 (persist_dir, nodes)
1193 macro_rules! open_channel {
1194 ($node_a: expr, $node_b: expr, $channel_value: expr) => {{
1195 begin_open_channel!($node_a, $node_b, $channel_value);
1196 let events = $node_a.node.get_and_clear_pending_events();
1197 assert_eq!(events.len(), 1);
1198 let (temporary_channel_id, tx) = handle_funding_generation_ready!(events[0], $channel_value);
1199 $node_a.node.funding_transaction_generated(&temporary_channel_id, &$node_b.node.get_our_node_id(), tx.clone()).unwrap();
1200 $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()));
1201 get_event!($node_b, Event::ChannelPending);
1202 $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()));
1203 get_event!($node_a, Event::ChannelPending);
1208 macro_rules! begin_open_channel {
1209 ($node_a: expr, $node_b: expr, $channel_value: expr) => {{
1210 $node_a.node.create_channel($node_b.node.get_our_node_id(), $channel_value, 100, 42, None).unwrap();
1211 $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()));
1212 $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()));
1216 macro_rules! handle_funding_generation_ready {
1217 ($event: expr, $channel_value: expr) => {{
1219 Event::FundingGenerationReady { temporary_channel_id, channel_value_satoshis, ref output_script, user_channel_id, .. } => {
1220 assert_eq!(channel_value_satoshis, $channel_value);
1221 assert_eq!(user_channel_id, 42);
1223 let tx = Transaction { version: 1 as i32, lock_time: PackedLockTime(0), input: Vec::new(), output: vec![TxOut {
1224 value: channel_value_satoshis, script_pubkey: output_script.clone(),
1226 (temporary_channel_id, tx)
1228 _ => panic!("Unexpected event"),
1233 fn confirm_transaction_depth(node: &mut Node, tx: &Transaction, depth: u32) {
1234 for i in 1..=depth {
1235 let prev_blockhash = node.best_block.block_hash();
1236 let height = node.best_block.height() + 1;
1237 let header = create_dummy_header(prev_blockhash, height);
1238 let txdata = vec![(0, tx)];
1239 node.best_block = BestBlock::new(header.block_hash(), height);
1242 node.node.transactions_confirmed(&header, &txdata, height);
1243 node.chain_monitor.transactions_confirmed(&header, &txdata, height);
1245 x if x == depth => {
1246 node.node.best_block_updated(&header, height);
1247 node.chain_monitor.best_block_updated(&header, height);
1253 fn confirm_transaction(node: &mut Node, tx: &Transaction) {
1254 confirm_transaction_depth(node, tx, ANTI_REORG_DELAY);
1258 fn test_background_processor() {
1259 // Test that when a new channel is created, the ChannelManager needs to be re-persisted with
1260 // updates. Also test that when new updates are available, the manager signals that it needs
1261 // re-persistence and is successfully re-persisted.
1262 let (persist_dir, nodes) = create_nodes(2, "test_background_processor");
1264 // Go through the channel creation process so that each node has something to persist. Since
1265 // open_channel consumes events, it must complete before starting BackgroundProcessor to
1266 // avoid a race with processing events.
1267 let tx = open_channel!(nodes[0], nodes[1], 100000);
1269 // Initiate the background processors to watch each node.
1270 let data_dir = nodes[0].persister.get_data_dir();
1271 let persister = Arc::new(Persister::new(data_dir));
1272 let event_handler = |_: _| {};
1273 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()));
1275 macro_rules! check_persisted_data {
1276 ($node: expr, $filepath: expr) => {
1277 let mut expected_bytes = Vec::new();
1279 expected_bytes.clear();
1280 match $node.write(&mut expected_bytes) {
1282 match std::fs::read($filepath) {
1284 if bytes == expected_bytes {
1293 Err(e) => panic!("Unexpected error: {}", e)
1299 // Check that the initial channel manager data is persisted as expected.
1300 let filepath = get_full_filepath(format!("{}_persister_0", &persist_dir), "manager".to_string());
1301 check_persisted_data!(nodes[0].node, filepath.clone());
1304 if !nodes[0].node.get_persistence_condvar_value() { break }
1307 // Force-close the channel.
1308 nodes[0].node.force_close_broadcasting_latest_txn(&OutPoint { txid: tx.txid(), index: 0 }.to_channel_id(), &nodes[1].node.get_our_node_id()).unwrap();
1310 // Check that the force-close updates are persisted.
1311 check_persisted_data!(nodes[0].node, filepath.clone());
1313 if !nodes[0].node.get_persistence_condvar_value() { break }
1316 // Check network graph is persisted
1317 let filepath = get_full_filepath(format!("{}_persister_0", &persist_dir), "network_graph".to_string());
1318 check_persisted_data!(nodes[0].network_graph, filepath.clone());
1320 // Check scorer is persisted
1321 let filepath = get_full_filepath(format!("{}_persister_0", &persist_dir), "scorer".to_string());
1322 check_persisted_data!(nodes[0].scorer, filepath.clone());
1324 if !std::thread::panicking() {
1325 bg_processor.stop().unwrap();
1330 fn test_timer_tick_called() {
1331 // Test that `ChannelManager::timer_tick_occurred` is called every `FRESHNESS_TIMER`,
1332 // `ChainMonitor::rebroadcast_pending_claims` is called every `REBROADCAST_TIMER`, and
1333 // `PeerManager::timer_tick_occurred` every `PING_TIMER`.
1334 let (_, nodes) = create_nodes(1, "test_timer_tick_called");
1335 let data_dir = nodes[0].persister.get_data_dir();
1336 let persister = Arc::new(Persister::new(data_dir));
1337 let event_handler = |_: _| {};
1338 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()));
1340 let log_entries = nodes[0].logger.lines.lock().unwrap();
1341 let desired_log_1 = "Calling ChannelManager's timer_tick_occurred".to_string();
1342 let desired_log_2 = "Calling PeerManager's timer_tick_occurred".to_string();
1343 let desired_log_3 = "Rebroadcasting monitor's pending claims".to_string();
1344 if log_entries.get(&("lightning_background_processor".to_string(), desired_log_1)).is_some() &&
1345 log_entries.get(&("lightning_background_processor".to_string(), desired_log_2)).is_some() &&
1346 log_entries.get(&("lightning_background_processor".to_string(), desired_log_3)).is_some() {
1351 if !std::thread::panicking() {
1352 bg_processor.stop().unwrap();
1357 fn test_channel_manager_persist_error() {
1358 // Test that if we encounter an error during manager persistence, the thread panics.
1359 let (_, nodes) = create_nodes(2, "test_persist_error");
1360 open_channel!(nodes[0], nodes[1], 100000);
1362 let data_dir = nodes[0].persister.get_data_dir();
1363 let persister = Arc::new(Persister::new(data_dir).with_manager_error(std::io::ErrorKind::Other, "test"));
1364 let event_handler = |_: _| {};
1365 let bg_processor = BackgroundProcessor::start(persister, event_handler, nodes[0].chain_monitor.clone(), nodes[0].node.clone(), nodes[0].no_gossip_sync(), nodes[0].peer_manager.clone(), nodes[0].logger.clone(), Some(nodes[0].scorer.clone()));
1366 match bg_processor.join() {
1367 Ok(_) => panic!("Expected error persisting manager"),
1369 assert_eq!(e.kind(), std::io::ErrorKind::Other);
1370 assert_eq!(e.get_ref().unwrap().to_string(), "test");
1376 #[cfg(feature = "futures")]
1377 async fn test_channel_manager_persist_error_async() {
1378 // Test that if we encounter an error during manager persistence, the thread panics.
1379 let (_, nodes) = create_nodes(2, "test_persist_error_sync");
1380 open_channel!(nodes[0], nodes[1], 100000);
1382 let data_dir = nodes[0].persister.get_data_dir();
1383 let persister = Arc::new(Persister::new(data_dir).with_manager_error(std::io::ErrorKind::Other, "test"));
1385 let bp_future = super::process_events_async(
1386 persister, |_: _| {async {}}, nodes[0].chain_monitor.clone(), nodes[0].node.clone(),
1387 nodes[0].rapid_gossip_sync(), nodes[0].peer_manager.clone(), nodes[0].logger.clone(),
1388 Some(nodes[0].scorer.clone()), move |dur: Duration| {
1389 Box::pin(async move {
1390 tokio::time::sleep(dur).await;
1395 match bp_future.await {
1396 Ok(_) => panic!("Expected error persisting manager"),
1398 assert_eq!(e.kind(), std::io::ErrorKind::Other);
1399 assert_eq!(e.get_ref().unwrap().to_string(), "test");
1405 fn test_network_graph_persist_error() {
1406 // Test that if we encounter an error during network graph persistence, an error gets returned.
1407 let (_, nodes) = create_nodes(2, "test_persist_network_graph_error");
1408 let data_dir = nodes[0].persister.get_data_dir();
1409 let persister = Arc::new(Persister::new(data_dir).with_graph_error(std::io::ErrorKind::Other, "test"));
1410 let event_handler = |_: _| {};
1411 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()));
1413 match bg_processor.stop() {
1414 Ok(_) => panic!("Expected error persisting network graph"),
1416 assert_eq!(e.kind(), std::io::ErrorKind::Other);
1417 assert_eq!(e.get_ref().unwrap().to_string(), "test");
1423 fn test_scorer_persist_error() {
1424 // Test that if we encounter an error during scorer persistence, an error gets returned.
1425 let (_, nodes) = create_nodes(2, "test_persist_scorer_error");
1426 let data_dir = nodes[0].persister.get_data_dir();
1427 let persister = Arc::new(Persister::new(data_dir).with_scorer_error(std::io::ErrorKind::Other, "test"));
1428 let event_handler = |_: _| {};
1429 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()));
1431 match bg_processor.stop() {
1432 Ok(_) => panic!("Expected error persisting scorer"),
1434 assert_eq!(e.kind(), std::io::ErrorKind::Other);
1435 assert_eq!(e.get_ref().unwrap().to_string(), "test");
1441 fn test_background_event_handling() {
1442 let (_, mut nodes) = create_nodes(2, "test_background_event_handling");
1443 let channel_value = 100000;
1444 let data_dir = nodes[0].persister.get_data_dir();
1445 let persister = Arc::new(Persister::new(data_dir.clone()));
1447 // Set up a background event handler for FundingGenerationReady events.
1448 let (funding_generation_send, funding_generation_recv) = std::sync::mpsc::sync_channel(1);
1449 let (channel_pending_send, channel_pending_recv) = std::sync::mpsc::sync_channel(1);
1450 let event_handler = move |event: Event| match event {
1451 Event::FundingGenerationReady { .. } => funding_generation_send.send(handle_funding_generation_ready!(event, channel_value)).unwrap(),
1452 Event::ChannelPending { .. } => channel_pending_send.send(()).unwrap(),
1453 Event::ChannelReady { .. } => {},
1454 _ => panic!("Unexpected event: {:?}", event),
1457 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()));
1459 // Open a channel and check that the FundingGenerationReady event was handled.
1460 begin_open_channel!(nodes[0], nodes[1], channel_value);
1461 let (temporary_channel_id, funding_tx) = funding_generation_recv
1462 .recv_timeout(Duration::from_secs(EVENT_DEADLINE))
1463 .expect("FundingGenerationReady not handled within deadline");
1464 nodes[0].node.funding_transaction_generated(&temporary_channel_id, &nodes[1].node.get_our_node_id(), funding_tx.clone()).unwrap();
1465 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()));
1466 get_event!(nodes[1], Event::ChannelPending);
1467 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()));
1468 let _ = channel_pending_recv.recv_timeout(Duration::from_secs(EVENT_DEADLINE))
1469 .expect("ChannelPending not handled within deadline");
1471 // Confirm the funding transaction.
1472 confirm_transaction(&mut nodes[0], &funding_tx);
1473 let as_funding = get_event_msg!(nodes[0], MessageSendEvent::SendChannelReady, nodes[1].node.get_our_node_id());
1474 confirm_transaction(&mut nodes[1], &funding_tx);
1475 let bs_funding = get_event_msg!(nodes[1], MessageSendEvent::SendChannelReady, nodes[0].node.get_our_node_id());
1476 nodes[0].node.handle_channel_ready(&nodes[1].node.get_our_node_id(), &bs_funding);
1477 let _as_channel_update = get_event_msg!(nodes[0], MessageSendEvent::SendChannelUpdate, nodes[1].node.get_our_node_id());
1478 nodes[1].node.handle_channel_ready(&nodes[0].node.get_our_node_id(), &as_funding);
1479 let _bs_channel_update = get_event_msg!(nodes[1], MessageSendEvent::SendChannelUpdate, nodes[0].node.get_our_node_id());
1481 if !std::thread::panicking() {
1482 bg_processor.stop().unwrap();
1485 // Set up a background event handler for SpendableOutputs events.
1486 let (sender, receiver) = std::sync::mpsc::sync_channel(1);
1487 let event_handler = move |event: Event| match event {
1488 Event::SpendableOutputs { .. } => sender.send(event).unwrap(),
1489 Event::ChannelReady { .. } => {},
1490 Event::ChannelClosed { .. } => {},
1491 _ => panic!("Unexpected event: {:?}", event),
1493 let persister = Arc::new(Persister::new(data_dir));
1494 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()));
1496 // Force close the channel and check that the SpendableOutputs event was handled.
1497 nodes[0].node.force_close_broadcasting_latest_txn(&nodes[0].node.list_channels()[0].channel_id, &nodes[1].node.get_our_node_id()).unwrap();
1498 let commitment_tx = nodes[0].tx_broadcaster.txn_broadcasted.lock().unwrap().pop().unwrap();
1499 confirm_transaction_depth(&mut nodes[0], &commitment_tx, BREAKDOWN_TIMEOUT as u32);
1501 let event = receiver
1502 .recv_timeout(Duration::from_secs(EVENT_DEADLINE))
1503 .expect("Events not handled within deadline");
1505 Event::SpendableOutputs { .. } => {},
1506 _ => panic!("Unexpected event: {:?}", event),
1509 if !std::thread::panicking() {
1510 bg_processor.stop().unwrap();
1515 fn test_scorer_persistence() {
1516 let (_, nodes) = create_nodes(2, "test_scorer_persistence");
1517 let data_dir = nodes[0].persister.get_data_dir();
1518 let persister = Arc::new(Persister::new(data_dir));
1519 let event_handler = |_: _| {};
1520 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()));
1523 let log_entries = nodes[0].logger.lines.lock().unwrap();
1524 let expected_log = "Persisting scorer".to_string();
1525 if log_entries.get(&("lightning_background_processor".to_string(), expected_log)).is_some() {
1530 if !std::thread::panicking() {
1531 bg_processor.stop().unwrap();
1535 macro_rules! do_test_not_pruning_network_graph_until_graph_sync_completion {
1536 ($nodes: expr, $receive: expr, $sleep: expr) => {
1537 let features = ChannelFeatures::empty();
1538 $nodes[0].network_graph.add_channel_from_partial_announcement(
1539 42, 53, features, $nodes[0].node.get_our_node_id(), $nodes[1].node.get_our_node_id()
1540 ).expect("Failed to update channel from partial announcement");
1541 let original_graph_description = $nodes[0].network_graph.to_string();
1542 assert!(original_graph_description.contains("42: features: 0000, node_one:"));
1543 assert_eq!($nodes[0].network_graph.read_only().channels().len(), 1);
1547 let log_entries = $nodes[0].logger.lines.lock().unwrap();
1548 let loop_counter = "Calling ChannelManager's timer_tick_occurred".to_string();
1549 if *log_entries.get(&("lightning_background_processor".to_string(), loop_counter))
1552 // Wait until the loop has gone around at least twice.
1557 let initialization_input = vec![
1558 76, 68, 75, 1, 111, 226, 140, 10, 182, 241, 179, 114, 193, 166, 162, 70, 174, 99, 247,
1559 79, 147, 30, 131, 101, 225, 90, 8, 156, 104, 214, 25, 0, 0, 0, 0, 0, 97, 227, 98, 218,
1560 0, 0, 0, 4, 2, 22, 7, 207, 206, 25, 164, 197, 231, 230, 231, 56, 102, 61, 250, 251,
1561 187, 172, 38, 46, 79, 247, 108, 44, 155, 48, 219, 238, 252, 53, 192, 6, 67, 2, 36, 125,
1562 157, 176, 223, 175, 234, 116, 94, 248, 201, 225, 97, 235, 50, 47, 115, 172, 63, 136,
1563 88, 216, 115, 11, 111, 217, 114, 84, 116, 124, 231, 107, 2, 158, 1, 242, 121, 152, 106,
1564 204, 131, 186, 35, 93, 70, 216, 10, 237, 224, 183, 89, 95, 65, 3, 83, 185, 58, 138,
1565 181, 64, 187, 103, 127, 68, 50, 2, 201, 19, 17, 138, 136, 149, 185, 226, 156, 137, 175,
1566 110, 32, 237, 0, 217, 90, 31, 100, 228, 149, 46, 219, 175, 168, 77, 4, 143, 38, 128,
1567 76, 97, 0, 0, 0, 2, 0, 0, 255, 8, 153, 192, 0, 2, 27, 0, 0, 0, 1, 0, 0, 255, 2, 68,
1568 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,
1569 0, 0, 0, 1, 0, 0, 0, 0, 58, 85, 116, 216, 255, 8, 153, 192, 0, 2, 27, 0, 0, 25, 0, 0,
1570 0, 1, 0, 0, 0, 125, 255, 2, 68, 226, 0, 6, 11, 0, 1, 5, 0, 0, 0, 0, 29, 129, 25, 192,
1572 $nodes[0].rapid_gossip_sync.update_network_graph_no_std(&initialization_input[..], Some(1642291930)).unwrap();
1574 // this should have added two channels and pruned the previous one.
1575 assert_eq!($nodes[0].network_graph.read_only().channels().len(), 2);
1577 $receive.expect("Network graph not pruned within deadline");
1579 // all channels should now be pruned
1580 assert_eq!($nodes[0].network_graph.read_only().channels().len(), 0);
1585 fn test_not_pruning_network_graph_until_graph_sync_completion() {
1586 let (sender, receiver) = std::sync::mpsc::sync_channel(1);
1588 let (_, nodes) = create_nodes(2, "test_not_pruning_network_graph_until_graph_sync_completion");
1589 let data_dir = nodes[0].persister.get_data_dir();
1590 let persister = Arc::new(Persister::new(data_dir).with_graph_persistence_notifier(sender));
1592 let event_handler = |_: _| {};
1593 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()));
1595 do_test_not_pruning_network_graph_until_graph_sync_completion!(nodes,
1596 receiver.recv_timeout(Duration::from_secs(super::FIRST_NETWORK_PRUNE_TIMER * 5)),
1597 std::thread::sleep(Duration::from_millis(1)));
1599 background_processor.stop().unwrap();
1603 #[cfg(feature = "futures")]
1604 async fn test_not_pruning_network_graph_until_graph_sync_completion_async() {
1605 let (sender, receiver) = std::sync::mpsc::sync_channel(1);
1607 let (_, nodes) = create_nodes(2, "test_not_pruning_network_graph_until_graph_sync_completion_async");
1608 let data_dir = nodes[0].persister.get_data_dir();
1609 let persister = Arc::new(Persister::new(data_dir).with_graph_persistence_notifier(sender));
1611 let (exit_sender, exit_receiver) = tokio::sync::watch::channel(());
1612 let bp_future = super::process_events_async(
1613 persister, |_: _| {async {}}, nodes[0].chain_monitor.clone(), nodes[0].node.clone(),
1614 nodes[0].rapid_gossip_sync(), nodes[0].peer_manager.clone(), nodes[0].logger.clone(),
1615 Some(nodes[0].scorer.clone()), move |dur: Duration| {
1616 let mut exit_receiver = exit_receiver.clone();
1617 Box::pin(async move {
1619 _ = tokio::time::sleep(dur) => false,
1620 _ = exit_receiver.changed() => true,
1626 let t1 = tokio::spawn(bp_future);
1627 let t2 = tokio::spawn(async move {
1628 do_test_not_pruning_network_graph_until_graph_sync_completion!(nodes, {
1631 tokio::time::sleep(Duration::from_secs(super::FIRST_NETWORK_PRUNE_TIMER)).await;
1632 if let Ok(()) = receiver.try_recv() { break Ok::<(), ()>(()); }
1636 }, tokio::time::sleep(Duration::from_millis(1)).await);
1637 exit_sender.send(()).unwrap();
1639 let (r1, r2) = tokio::join!(t1, t2);
1640 r1.unwrap().unwrap();
1644 macro_rules! do_test_payment_path_scoring {
1645 ($nodes: expr, $receive: expr) => {
1646 // Ensure that we update the scorer when relevant events are processed. In this case, we ensure
1647 // that we update the scorer upon a payment path succeeding (note that the channel must be
1648 // public or else we won't score it).
1649 // A background event handler for FundingGenerationReady events must be hooked up to a
1650 // running background processor.
1651 let scored_scid = 4242;
1652 let secp_ctx = Secp256k1::new();
1653 let node_1_privkey = SecretKey::from_slice(&[42; 32]).unwrap();
1654 let node_1_id = PublicKey::from_secret_key(&secp_ctx, &node_1_privkey);
1656 let path = Path { hops: vec![RouteHop {
1658 node_features: NodeFeatures::empty(),
1659 short_channel_id: scored_scid,
1660 channel_features: ChannelFeatures::empty(),
1662 cltv_expiry_delta: MIN_CLTV_EXPIRY_DELTA as u32,
1663 }], blinded_tail: None };
1665 $nodes[0].scorer.lock().unwrap().expect(TestResult::PaymentFailure { path: path.clone(), short_channel_id: scored_scid });
1666 $nodes[0].node.push_pending_event(Event::PaymentPathFailed {
1668 payment_hash: PaymentHash([42; 32]),
1669 payment_failed_permanently: false,
1670 failure: PathFailure::OnPath { network_update: None },
1672 short_channel_id: Some(scored_scid),
1674 let event = $receive.expect("PaymentPathFailed not handled within deadline");
1676 Event::PaymentPathFailed { .. } => {},
1677 _ => panic!("Unexpected event"),
1680 // Ensure we'll score payments that were explicitly failed back by the destination as
1682 $nodes[0].scorer.lock().unwrap().expect(TestResult::ProbeSuccess { path: path.clone() });
1683 $nodes[0].node.push_pending_event(Event::PaymentPathFailed {
1685 payment_hash: PaymentHash([42; 32]),
1686 payment_failed_permanently: true,
1687 failure: PathFailure::OnPath { network_update: None },
1689 short_channel_id: None,
1691 let event = $receive.expect("PaymentPathFailed not handled within deadline");
1693 Event::PaymentPathFailed { .. } => {},
1694 _ => panic!("Unexpected event"),
1697 $nodes[0].scorer.lock().unwrap().expect(TestResult::PaymentSuccess { path: path.clone() });
1698 $nodes[0].node.push_pending_event(Event::PaymentPathSuccessful {
1699 payment_id: PaymentId([42; 32]),
1703 let event = $receive.expect("PaymentPathSuccessful not handled within deadline");
1705 Event::PaymentPathSuccessful { .. } => {},
1706 _ => panic!("Unexpected event"),
1709 $nodes[0].scorer.lock().unwrap().expect(TestResult::ProbeSuccess { path: path.clone() });
1710 $nodes[0].node.push_pending_event(Event::ProbeSuccessful {
1711 payment_id: PaymentId([42; 32]),
1712 payment_hash: PaymentHash([42; 32]),
1715 let event = $receive.expect("ProbeSuccessful not handled within deadline");
1717 Event::ProbeSuccessful { .. } => {},
1718 _ => panic!("Unexpected event"),
1721 $nodes[0].scorer.lock().unwrap().expect(TestResult::ProbeFailure { path: path.clone() });
1722 $nodes[0].node.push_pending_event(Event::ProbeFailed {
1723 payment_id: PaymentId([42; 32]),
1724 payment_hash: PaymentHash([42; 32]),
1726 short_channel_id: Some(scored_scid),
1728 let event = $receive.expect("ProbeFailure not handled within deadline");
1730 Event::ProbeFailed { .. } => {},
1731 _ => panic!("Unexpected event"),
1737 fn test_payment_path_scoring() {
1738 let (sender, receiver) = std::sync::mpsc::sync_channel(1);
1739 let event_handler = move |event: Event| match event {
1740 Event::PaymentPathFailed { .. } => sender.send(event).unwrap(),
1741 Event::PaymentPathSuccessful { .. } => sender.send(event).unwrap(),
1742 Event::ProbeSuccessful { .. } => sender.send(event).unwrap(),
1743 Event::ProbeFailed { .. } => sender.send(event).unwrap(),
1744 _ => panic!("Unexpected event: {:?}", event),
1747 let (_, nodes) = create_nodes(1, "test_payment_path_scoring");
1748 let data_dir = nodes[0].persister.get_data_dir();
1749 let persister = Arc::new(Persister::new(data_dir));
1750 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()));
1752 do_test_payment_path_scoring!(nodes, receiver.recv_timeout(Duration::from_secs(EVENT_DEADLINE)));
1754 if !std::thread::panicking() {
1755 bg_processor.stop().unwrap();
1758 let log_entries = nodes[0].logger.lines.lock().unwrap();
1759 let expected_log = "Persisting scorer after update".to_string();
1760 assert_eq!(*log_entries.get(&("lightning_background_processor".to_string(), expected_log)).unwrap(), 5);
1764 #[cfg(feature = "futures")]
1765 async fn test_payment_path_scoring_async() {
1766 let (sender, mut receiver) = tokio::sync::mpsc::channel(1);
1767 let event_handler = move |event: Event| {
1768 let sender_ref = sender.clone();
1771 Event::PaymentPathFailed { .. } => { sender_ref.send(event).await.unwrap() },
1772 Event::PaymentPathSuccessful { .. } => { sender_ref.send(event).await.unwrap() },
1773 Event::ProbeSuccessful { .. } => { sender_ref.send(event).await.unwrap() },
1774 Event::ProbeFailed { .. } => { sender_ref.send(event).await.unwrap() },
1775 _ => panic!("Unexpected event: {:?}", event),
1780 let (_, nodes) = create_nodes(1, "test_payment_path_scoring_async");
1781 let data_dir = nodes[0].persister.get_data_dir();
1782 let persister = Arc::new(Persister::new(data_dir));
1784 let (exit_sender, exit_receiver) = tokio::sync::watch::channel(());
1786 let bp_future = super::process_events_async(
1787 persister, event_handler, nodes[0].chain_monitor.clone(), nodes[0].node.clone(),
1788 nodes[0].no_gossip_sync(), nodes[0].peer_manager.clone(), nodes[0].logger.clone(),
1789 Some(nodes[0].scorer.clone()), move |dur: Duration| {
1790 let mut exit_receiver = exit_receiver.clone();
1791 Box::pin(async move {
1793 _ = tokio::time::sleep(dur) => false,
1794 _ = exit_receiver.changed() => true,
1799 let t1 = tokio::spawn(bp_future);
1800 let t2 = tokio::spawn(async move {
1801 do_test_payment_path_scoring!(nodes, receiver.recv().await);
1802 exit_sender.send(()).unwrap();
1804 let log_entries = nodes[0].logger.lines.lock().unwrap();
1805 let expected_log = "Persisting scorer after update".to_string();
1806 assert_eq!(*log_entries.get(&("lightning_background_processor".to_string(), expected_log)).unwrap(), 5);
1809 let (r1, r2) = tokio::join!(t1, t2);
1810 r1.unwrap().unwrap();