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::chain::keysinterface::{EntropySource, NodeSigner, SignerProvider};
29 use lightning::events::{Event, PathFailure};
30 #[cfg(feature = "std")]
31 use lightning::events::{EventHandler, EventsProvider};
32 use lightning::ln::channelmanager::ChannelManager;
33 use lightning::ln::msgs::{ChannelMessageHandler, OnionMessageHandler, RoutingMessageHandler};
34 use lightning::ln::peer_handler::{CustomMessageHandler, PeerManager, SocketDescriptor};
35 use lightning::routing::gossip::{NetworkGraph, P2PGossipSync};
36 use lightning::routing::utxo::UtxoLookup;
37 use lightning::routing::router::Router;
38 use lightning::routing::scoring::{Score, 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;
46 use core::time::Duration;
48 #[cfg(feature = "std")]
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;
57 #[cfg(not(feature = "std"))]
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`]).
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.
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.
82 /// [`ChannelMonitor`]: lightning::chain::channelmonitor::ChannelMonitor
83 /// [`Event`]: lightning::events::Event
84 #[cfg(feature = "std")]
85 #[must_use = "BackgroundProcessor will immediately stop on drop. It should be stored until shutdown."]
86 pub struct BackgroundProcessor {
87 stop_thread: Arc<AtomicBool>,
88 thread_handle: Option<JoinHandle<Result<(), std::io::Error>>>,
92 const FRESHNESS_TIMER: u64 = 60;
94 const FRESHNESS_TIMER: u64 = 1;
96 #[cfg(all(not(test), not(debug_assertions)))]
97 const PING_TIMER: u64 = 10;
98 /// Signature operations take a lot longer without compiler optimisations.
99 /// Increasing the ping timer allows for this but slower devices will be disconnected if the
100 /// timeout is reached.
101 #[cfg(all(not(test), debug_assertions))]
102 const PING_TIMER: u64 = 30;
104 const PING_TIMER: u64 = 1;
106 /// Prune the network graph of stale entries hourly.
107 const NETWORK_PRUNE_TIMER: u64 = 60 * 60;
110 const SCORER_PERSIST_TIMER: u64 = 30;
112 const SCORER_PERSIST_TIMER: u64 = 1;
115 const FIRST_NETWORK_PRUNE_TIMER: u64 = 60;
117 const FIRST_NETWORK_PRUNE_TIMER: u64 = 1;
120 const REBROADCAST_TIMER: u64 = 30;
122 const REBROADCAST_TIMER: u64 = 1;
124 #[cfg(feature = "futures")]
125 /// core::cmp::min is not currently const, so we define a trivial (and equivalent) replacement
126 const fn min_u64(a: u64, b: u64) -> u64 { if a < b { a } else { b } }
127 #[cfg(feature = "futures")]
128 const FASTEST_TIMER: u64 = min_u64(min_u64(FRESHNESS_TIMER, PING_TIMER),
129 min_u64(SCORER_PERSIST_TIMER, min_u64(FIRST_NETWORK_PRUNE_TIMER, REBROADCAST_TIMER)));
131 /// Either [`P2PGossipSync`] or [`RapidGossipSync`].
133 P: Deref<Target = P2PGossipSync<G, U, L>>,
134 R: Deref<Target = RapidGossipSync<G, L>>,
135 G: Deref<Target = NetworkGraph<L>>,
139 where U::Target: UtxoLookup, L::Target: Logger {
140 /// Gossip sync via the lightning peer-to-peer network as defined by BOLT 7.
142 /// Rapid gossip sync from a trusted server.
149 P: Deref<Target = P2PGossipSync<G, U, L>>,
150 R: Deref<Target = RapidGossipSync<G, L>>,
151 G: Deref<Target = NetworkGraph<L>>,
154 > GossipSync<P, R, G, U, L>
155 where U::Target: UtxoLookup, L::Target: Logger {
156 fn network_graph(&self) -> Option<&G> {
158 GossipSync::P2P(gossip_sync) => Some(gossip_sync.network_graph()),
159 GossipSync::Rapid(gossip_sync) => Some(gossip_sync.network_graph()),
160 GossipSync::None => None,
164 fn prunable_network_graph(&self) -> Option<&G> {
166 GossipSync::P2P(gossip_sync) => Some(gossip_sync.network_graph()),
167 GossipSync::Rapid(gossip_sync) => {
168 if gossip_sync.is_initial_sync_complete() {
169 Some(gossip_sync.network_graph())
174 GossipSync::None => None,
179 /// This is not exported to bindings users as the bindings concretize everything and have constructors for us
180 impl<P: Deref<Target = P2PGossipSync<G, U, L>>, G: Deref<Target = NetworkGraph<L>>, U: Deref, L: Deref>
181 GossipSync<P, &RapidGossipSync<G, L>, G, U, L>
183 U::Target: UtxoLookup,
186 /// Initializes a new [`GossipSync::P2P`] variant.
187 pub fn p2p(gossip_sync: P) -> Self {
188 GossipSync::P2P(gossip_sync)
192 /// This is not exported to bindings users as the bindings concretize everything and have constructors for us
193 impl<'a, R: Deref<Target = RapidGossipSync<G, L>>, G: Deref<Target = NetworkGraph<L>>, L: Deref>
195 &P2PGossipSync<G, &'a (dyn UtxoLookup + Send + Sync), L>,
198 &'a (dyn UtxoLookup + Send + Sync),
204 /// Initializes a new [`GossipSync::Rapid`] variant.
205 pub fn rapid(gossip_sync: R) -> Self {
206 GossipSync::Rapid(gossip_sync)
210 /// This is not exported to bindings users as the bindings concretize everything and have constructors for us
213 &P2PGossipSync<&'a NetworkGraph<L>, &'a (dyn UtxoLookup + Send + Sync), L>,
214 &RapidGossipSync<&'a NetworkGraph<L>, L>,
216 &'a (dyn UtxoLookup + Send + Sync),
222 /// Initializes a new [`GossipSync::None`] variant.
223 pub fn none() -> Self {
228 fn handle_network_graph_update<L: Deref>(
229 network_graph: &NetworkGraph<L>, event: &Event
230 ) where L::Target: Logger {
231 if let Event::PaymentPathFailed {
232 failure: PathFailure::OnPath { network_update: Some(ref upd) }, .. } = event
234 network_graph.handle_network_update(upd);
238 fn update_scorer<'a, S: 'static + Deref<Target = SC> + Send + Sync, SC: 'a + WriteableScore<'a>>(
239 scorer: &'a S, event: &Event
241 let mut score = scorer.lock();
243 Event::PaymentPathFailed { ref path, short_channel_id: Some(scid), .. } => {
244 score.payment_path_failed(path, *scid);
246 Event::PaymentPathFailed { ref path, payment_failed_permanently: true, .. } => {
247 // Reached if the destination explicitly failed it back. We treat this as a successful probe
248 // because the payment made it all the way to the destination with sufficient liquidity.
249 score.probe_successful(path);
251 Event::PaymentPathSuccessful { path, .. } => {
252 score.payment_path_successful(path);
254 Event::ProbeSuccessful { path, .. } => {
255 score.probe_successful(path);
257 Event::ProbeFailed { path, short_channel_id: Some(scid), .. } => {
258 score.probe_failed(path, *scid);
264 macro_rules! define_run_body {
265 ($persister: ident, $chain_monitor: ident, $process_chain_monitor_events: expr,
266 $channel_manager: ident, $process_channel_manager_events: expr,
267 $gossip_sync: ident, $peer_manager: ident, $logger: ident, $scorer: ident,
268 $loop_exit_check: expr, $await: expr, $get_timer: expr, $timer_elapsed: expr,
269 $check_slow_await: expr)
271 log_trace!($logger, "Calling ChannelManager's timer_tick_occurred on startup");
272 $channel_manager.timer_tick_occurred();
273 log_trace!($logger, "Rebroadcasting monitor's pending claims on startup");
274 $chain_monitor.rebroadcast_pending_claims();
276 let mut last_freshness_call = $get_timer(FRESHNESS_TIMER);
277 let mut last_ping_call = $get_timer(PING_TIMER);
278 let mut last_prune_call = $get_timer(FIRST_NETWORK_PRUNE_TIMER);
279 let mut last_scorer_persist_call = $get_timer(SCORER_PERSIST_TIMER);
280 let mut last_rebroadcast_call = $get_timer(REBROADCAST_TIMER);
281 let mut have_pruned = false;
284 $process_channel_manager_events;
285 $process_chain_monitor_events;
287 // Note that the PeerManager::process_events may block on ChannelManager's locks,
288 // hence it comes last here. When the ChannelManager finishes whatever it's doing,
289 // we want to ensure we get into `persist_manager` as quickly as we can, especially
290 // without running the normal event processing above and handing events to users.
292 // Specifically, on an *extremely* slow machine, we may see ChannelManager start
293 // processing a message effectively at any point during this loop. In order to
294 // minimize the time between such processing completing and persisting the updated
295 // ChannelManager, we want to minimize methods blocking on a ChannelManager
296 // generally, and as a fallback place such blocking only immediately before
298 $peer_manager.process_events();
300 // Exit the loop if the background processor was requested to stop.
301 if $loop_exit_check {
302 log_trace!($logger, "Terminating background processor.");
306 // We wait up to 100ms, but track how long it takes to detect being put to sleep,
307 // see `await_start`'s use below.
308 let mut await_start = None;
309 if $check_slow_await { await_start = Some($get_timer(1)); }
310 let updates_available = $await;
311 let await_slow = if $check_slow_await { $timer_elapsed(&mut await_start.unwrap(), 1) } else { false };
313 // Exit the loop if the background processor was requested to stop.
314 if $loop_exit_check {
315 log_trace!($logger, "Terminating background processor.");
319 if updates_available {
320 log_trace!($logger, "Persisting ChannelManager...");
321 $persister.persist_manager(&*$channel_manager)?;
322 log_trace!($logger, "Done persisting ChannelManager.");
324 if $timer_elapsed(&mut last_freshness_call, FRESHNESS_TIMER) {
325 log_trace!($logger, "Calling ChannelManager's timer_tick_occurred");
326 $channel_manager.timer_tick_occurred();
327 last_freshness_call = $get_timer(FRESHNESS_TIMER);
330 // On various platforms, we may be starved of CPU cycles for several reasons.
331 // E.g. on iOS, if we've been in the background, we will be entirely paused.
332 // Similarly, if we're on a desktop platform and the device has been asleep, we
333 // may not get any cycles.
334 // We detect this by checking if our max-100ms-sleep, above, ran longer than a
335 // full second, at which point we assume sockets may have been killed (they
336 // appear to be at least on some platforms, even if it has only been a second).
337 // Note that we have to take care to not get here just because user event
338 // processing was slow at the top of the loop. For example, the sample client
339 // may call Bitcoin Core RPCs during event handling, which very often takes
340 // more than a handful of seconds to complete, and shouldn't disconnect all our
342 log_trace!($logger, "100ms sleep took more than a second, disconnecting peers.");
343 $peer_manager.disconnect_all_peers();
344 last_ping_call = $get_timer(PING_TIMER);
345 } else if $timer_elapsed(&mut last_ping_call, PING_TIMER) {
346 log_trace!($logger, "Calling PeerManager's timer_tick_occurred");
347 $peer_manager.timer_tick_occurred();
348 last_ping_call = $get_timer(PING_TIMER);
351 // Note that we want to run a graph prune once not long after startup before
352 // falling back to our usual hourly prunes. This avoids short-lived clients never
353 // pruning their network graph. We run once 60 seconds after startup before
354 // continuing our normal cadence. For RGS, since 60 seconds is likely too long,
355 // we prune after an initial sync completes.
356 let prune_timer = if have_pruned { NETWORK_PRUNE_TIMER } else { FIRST_NETWORK_PRUNE_TIMER };
357 let prune_timer_elapsed = $timer_elapsed(&mut last_prune_call, prune_timer);
358 let should_prune = match $gossip_sync {
359 GossipSync::Rapid(_) => !have_pruned || prune_timer_elapsed,
360 _ => prune_timer_elapsed,
363 // The network graph must not be pruned while rapid sync completion is pending
364 if let Some(network_graph) = $gossip_sync.prunable_network_graph() {
365 #[cfg(feature = "std")] {
366 log_trace!($logger, "Pruning and persisting network graph.");
367 network_graph.remove_stale_channels_and_tracking();
369 #[cfg(not(feature = "std"))] {
370 log_warn!($logger, "Not pruning network graph, consider enabling `std` or doing so manually with remove_stale_channels_and_tracking_with_time.");
371 log_trace!($logger, "Persisting network graph.");
374 if let Err(e) = $persister.persist_graph(network_graph) {
375 log_error!($logger, "Error: Failed to persist network graph, check your disk and permissions {}", e)
380 let prune_timer = if have_pruned { NETWORK_PRUNE_TIMER } else { FIRST_NETWORK_PRUNE_TIMER };
381 last_prune_call = $get_timer(prune_timer);
384 if $timer_elapsed(&mut last_scorer_persist_call, SCORER_PERSIST_TIMER) {
385 if let Some(ref scorer) = $scorer {
386 log_trace!($logger, "Persisting scorer");
387 if let Err(e) = $persister.persist_scorer(&scorer) {
388 log_error!($logger, "Error: Failed to persist scorer, check your disk and permissions {}", e)
391 last_scorer_persist_call = $get_timer(SCORER_PERSIST_TIMER);
394 if $timer_elapsed(&mut last_rebroadcast_call, REBROADCAST_TIMER) {
395 log_trace!($logger, "Rebroadcasting monitor's pending claims");
396 $chain_monitor.rebroadcast_pending_claims();
397 last_rebroadcast_call = $get_timer(REBROADCAST_TIMER);
401 // After we exit, ensure we persist the ChannelManager one final time - this avoids
402 // some races where users quit while channel updates were in-flight, with
403 // ChannelMonitor update(s) persisted without a corresponding ChannelManager update.
404 $persister.persist_manager(&*$channel_manager)?;
406 // Persist Scorer on exit
407 if let Some(ref scorer) = $scorer {
408 $persister.persist_scorer(&scorer)?;
411 // Persist NetworkGraph on exit
412 if let Some(network_graph) = $gossip_sync.network_graph() {
413 $persister.persist_graph(network_graph)?;
420 #[cfg(feature = "futures")]
421 pub(crate) mod futures_util {
422 use core::future::Future;
423 use core::task::{Poll, Waker, RawWaker, RawWakerVTable};
425 use core::marker::Unpin;
426 pub(crate) struct Selector<
427 A: Future<Output=()> + Unpin, B: Future<Output=()> + Unpin, C: Future<Output=bool> + Unpin
433 pub(crate) enum SelectorOutput {
438 A: Future<Output=()> + Unpin, B: Future<Output=()> + Unpin, C: Future<Output=bool> + Unpin
439 > Future for Selector<A, B, C> {
440 type Output = SelectorOutput;
441 fn poll(mut self: Pin<&mut Self>, ctx: &mut core::task::Context<'_>) -> Poll<SelectorOutput> {
442 match Pin::new(&mut self.a).poll(ctx) {
443 Poll::Ready(()) => { return Poll::Ready(SelectorOutput::A); },
446 match Pin::new(&mut self.b).poll(ctx) {
447 Poll::Ready(()) => { return Poll::Ready(SelectorOutput::B); },
450 match Pin::new(&mut self.c).poll(ctx) {
451 Poll::Ready(res) => { return Poll::Ready(SelectorOutput::C(res)); },
458 // If we want to poll a future without an async context to figure out if it has completed or
459 // not without awaiting, we need a Waker, which needs a vtable...we fill it with dummy values
460 // but sadly there's a good bit of boilerplate here.
461 fn dummy_waker_clone(_: *const ()) -> RawWaker { RawWaker::new(core::ptr::null(), &DUMMY_WAKER_VTABLE) }
462 fn dummy_waker_action(_: *const ()) { }
464 const DUMMY_WAKER_VTABLE: RawWakerVTable = RawWakerVTable::new(
465 dummy_waker_clone, dummy_waker_action, dummy_waker_action, dummy_waker_action);
466 pub(crate) fn dummy_waker() -> Waker { unsafe { Waker::from_raw(RawWaker::new(core::ptr::null(), &DUMMY_WAKER_VTABLE)) } }
468 #[cfg(feature = "futures")]
469 use futures_util::{Selector, SelectorOutput, dummy_waker};
470 #[cfg(feature = "futures")]
473 /// Processes background events in a future.
475 /// `sleeper` should return a future which completes in the given amount of time and returns a
476 /// boolean indicating whether the background processing should exit. Once `sleeper` returns a
477 /// future which outputs `true`, the loop will exit and this function's future will complete.
478 /// The `sleeper` future is free to return early after it has triggered the exit condition.
480 /// See [`BackgroundProcessor::start`] for information on which actions this handles.
482 /// Requires the `futures` feature. Note that while this method is available without the `std`
483 /// feature, doing so will skip calling [`NetworkGraph::remove_stale_channels_and_tracking`],
484 /// you should call [`NetworkGraph::remove_stale_channels_and_tracking_with_time`] regularly
485 /// manually instead.
487 /// The `mobile_interruptable_platform` flag should be set if we're currently running on a
488 /// mobile device, where we may need to check for interruption of the application regularly. If you
489 /// are unsure, you should set the flag, as the performance impact of it is minimal unless there
490 /// are hundreds or thousands of simultaneous process calls running.
492 /// For example, in order to process background events in a [Tokio](https://tokio.rs/) task, you
493 /// could setup `process_events_async` like this:
495 /// # struct MyPersister {}
496 /// # impl lightning::util::persist::KVStorePersister for MyPersister {
497 /// # fn persist<W: lightning::util::ser::Writeable>(&self, key: &str, object: &W) -> lightning::io::Result<()> { Ok(()) }
499 /// # struct MyEventHandler {}
500 /// # impl MyEventHandler {
501 /// # async fn handle_event(&self, _: lightning::events::Event) {}
503 /// # #[derive(Eq, PartialEq, Clone, Hash)]
504 /// # struct MySocketDescriptor {}
505 /// # impl lightning::ln::peer_handler::SocketDescriptor for MySocketDescriptor {
506 /// # fn send_data(&mut self, _data: &[u8], _resume_read: bool) -> usize { 0 }
507 /// # fn disconnect_socket(&mut self) {}
509 /// # use std::sync::{Arc, Mutex};
510 /// # use std::sync::atomic::{AtomicBool, Ordering};
511 /// # use lightning_background_processor::{process_events_async, GossipSync};
512 /// # type MyBroadcaster = dyn lightning::chain::chaininterface::BroadcasterInterface + Send + Sync;
513 /// # type MyFeeEstimator = dyn lightning::chain::chaininterface::FeeEstimator + Send + Sync;
514 /// # type MyNodeSigner = dyn lightning::chain::keysinterface::NodeSigner + Send + Sync;
515 /// # type MyUtxoLookup = dyn lightning::routing::utxo::UtxoLookup + Send + Sync;
516 /// # type MyFilter = dyn lightning::chain::Filter + Send + Sync;
517 /// # type MyLogger = dyn lightning::util::logger::Logger + Send + Sync;
518 /// # type MyChainMonitor = lightning::chain::chainmonitor::ChainMonitor<lightning::chain::keysinterface::InMemorySigner, Arc<MyFilter>, Arc<MyBroadcaster>, Arc<MyFeeEstimator>, Arc<MyLogger>, Arc<MyPersister>>;
519 /// # type MyPeerManager = lightning::ln::peer_handler::SimpleArcPeerManager<MySocketDescriptor, MyChainMonitor, MyBroadcaster, MyFeeEstimator, MyUtxoLookup, MyLogger>;
520 /// # type MyNetworkGraph = lightning::routing::gossip::NetworkGraph<Arc<MyLogger>>;
521 /// # type MyGossipSync = lightning::routing::gossip::P2PGossipSync<Arc<MyNetworkGraph>, Arc<MyUtxoLookup>, Arc<MyLogger>>;
522 /// # type MyChannelManager = lightning::ln::channelmanager::SimpleArcChannelManager<MyChainMonitor, MyBroadcaster, MyFeeEstimator, MyLogger>;
523 /// # type MyScorer = Mutex<lightning::routing::scoring::ProbabilisticScorer<Arc<MyNetworkGraph>, Arc<MyLogger>>>;
525 /// # 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>) {
526 /// let background_persister = Arc::clone(&my_persister);
527 /// let background_event_handler = Arc::clone(&my_event_handler);
528 /// let background_chain_mon = Arc::clone(&my_chain_monitor);
529 /// let background_chan_man = Arc::clone(&my_channel_manager);
530 /// let background_gossip_sync = GossipSync::p2p(Arc::clone(&my_gossip_sync));
531 /// let background_peer_man = Arc::clone(&my_peer_manager);
532 /// let background_logger = Arc::clone(&my_logger);
533 /// let background_scorer = Arc::clone(&my_scorer);
535 /// // Setup the sleeper.
536 /// let (stop_sender, stop_receiver) = tokio::sync::watch::channel(());
538 /// let sleeper = move |d| {
539 /// let mut receiver = stop_receiver.clone();
540 /// Box::pin(async move {
542 /// _ = tokio::time::sleep(d) => false,
543 /// _ = receiver.changed() => true,
548 /// let mobile_interruptable_platform = false;
550 /// let handle = tokio::spawn(async move {
551 /// process_events_async(
552 /// background_persister,
553 /// |e| background_event_handler.handle_event(e),
554 /// background_chain_mon,
555 /// background_chan_man,
556 /// background_gossip_sync,
557 /// background_peer_man,
558 /// background_logger,
559 /// Some(background_scorer),
561 /// mobile_interruptable_platform,
564 /// .expect("Failed to process events");
567 /// // Stop the background processing.
568 /// stop_sender.send(()).unwrap();
569 /// handle.await.unwrap();
572 #[cfg(feature = "futures")]
573 pub async fn process_events_async<
575 UL: 'static + Deref + Send + Sync,
576 CF: 'static + Deref + Send + Sync,
577 CW: 'static + Deref + Send + Sync,
578 T: 'static + Deref + Send + Sync,
579 ES: 'static + Deref + Send + Sync,
580 NS: 'static + Deref + Send + Sync,
581 SP: 'static + Deref + Send + Sync,
582 F: 'static + Deref + Send + Sync,
583 R: 'static + Deref + Send + Sync,
584 G: 'static + Deref<Target = NetworkGraph<L>> + Send + Sync,
585 L: 'static + Deref + Send + Sync,
586 P: 'static + Deref + Send + Sync,
587 Descriptor: 'static + SocketDescriptor + Send + Sync,
588 CMH: 'static + Deref + Send + Sync,
589 RMH: 'static + Deref + Send + Sync,
590 OMH: 'static + Deref + Send + Sync,
591 EventHandlerFuture: core::future::Future<Output = ()>,
592 EventHandler: Fn(Event) -> EventHandlerFuture,
593 PS: 'static + Deref + Send,
594 M: 'static + Deref<Target = ChainMonitor<<SP::Target as SignerProvider>::Signer, CF, T, F, L, P>> + Send + Sync,
595 CM: 'static + Deref<Target = ChannelManager<CW, T, ES, NS, SP, F, R, L>> + Send + Sync,
596 PGS: 'static + Deref<Target = P2PGossipSync<G, UL, L>> + Send + Sync,
597 RGS: 'static + Deref<Target = RapidGossipSync<G, L>> + Send,
598 UMH: 'static + Deref + Send + Sync,
599 PM: 'static + Deref<Target = PeerManager<Descriptor, CMH, RMH, OMH, L, UMH, NS>> + Send + Sync,
600 S: 'static + Deref<Target = SC> + Send + Sync,
601 SC: for<'b> WriteableScore<'b>,
602 SleepFuture: core::future::Future<Output = bool> + core::marker::Unpin,
603 Sleeper: Fn(Duration) -> SleepFuture
605 persister: PS, event_handler: EventHandler, chain_monitor: M, channel_manager: CM,
606 gossip_sync: GossipSync<PGS, RGS, G, UL, L>, peer_manager: PM, logger: L, scorer: Option<S>,
607 sleeper: Sleeper, mobile_interruptable_platform: bool,
608 ) -> Result<(), lightning::io::Error>
610 UL::Target: 'static + UtxoLookup,
611 CF::Target: 'static + chain::Filter,
612 CW::Target: 'static + chain::Watch<<SP::Target as SignerProvider>::Signer>,
613 T::Target: 'static + BroadcasterInterface,
614 ES::Target: 'static + EntropySource,
615 NS::Target: 'static + NodeSigner,
616 SP::Target: 'static + SignerProvider,
617 F::Target: 'static + FeeEstimator,
618 R::Target: 'static + Router,
619 L::Target: 'static + Logger,
620 P::Target: 'static + Persist<<SP::Target as SignerProvider>::Signer>,
621 CMH::Target: 'static + ChannelMessageHandler,
622 OMH::Target: 'static + OnionMessageHandler,
623 RMH::Target: 'static + RoutingMessageHandler,
624 UMH::Target: 'static + CustomMessageHandler,
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;
633 if let Some(network_graph) = network_graph {
634 handle_network_graph_update(network_graph, &event)
636 if let Some(ref scorer) = scorer {
637 update_scorer(scorer, &event);
639 event_handler(event).await;
642 define_run_body!(persister,
643 chain_monitor, chain_monitor.process_pending_events_async(async_event_handler).await,
644 channel_manager, channel_manager.process_pending_events_async(async_event_handler).await,
645 gossip_sync, peer_manager, logger, scorer, should_break, {
647 a: channel_manager.get_persistable_update_future(),
648 b: chain_monitor.get_update_future(),
649 c: sleeper(if mobile_interruptable_platform { Duration::from_millis(100) } else { Duration::from_secs(FASTEST_TIMER) }),
652 SelectorOutput::A => true,
653 SelectorOutput::B => false,
654 SelectorOutput::C(exit) => {
659 }, |t| sleeper(Duration::from_secs(t)),
660 |fut: &mut SleepFuture, _| {
661 let mut waker = dummy_waker();
662 let mut ctx = task::Context::from_waker(&mut waker);
663 match core::pin::Pin::new(fut).poll(&mut ctx) {
664 task::Poll::Ready(exit) => { should_break = exit; true },
665 task::Poll::Pending => false,
667 }, mobile_interruptable_platform)
670 #[cfg(feature = "std")]
671 impl BackgroundProcessor {
672 /// Start a background thread that takes care of responsibilities enumerated in the [top-level
675 /// The thread runs indefinitely unless the object is dropped, [`stop`] is called, or
676 /// [`Persister::persist_manager`] returns an error. In case of an error, the error is retrieved by calling
677 /// either [`join`] or [`stop`].
679 /// # Data Persistence
681 /// [`Persister::persist_manager`] is responsible for writing out the [`ChannelManager`] to disk, and/or
682 /// uploading to one or more backup services. See [`ChannelManager::write`] for writing out a
683 /// [`ChannelManager`]. See the `lightning-persister` crate for LDK's
684 /// provided implementation.
686 /// [`Persister::persist_graph`] is responsible for writing out the [`NetworkGraph`] to disk, if
687 /// [`GossipSync`] is supplied. See [`NetworkGraph::write`] for writing out a [`NetworkGraph`].
688 /// See the `lightning-persister` crate for LDK's provided implementation.
690 /// Typically, users should either implement [`Persister::persist_manager`] to never return an
691 /// error or call [`join`] and handle any error that may arise. For the latter case,
692 /// `BackgroundProcessor` must be restarted by calling `start` again after handling the error.
696 /// `event_handler` is responsible for handling events that users should be notified of (e.g.,
697 /// payment failed). [`BackgroundProcessor`] may decorate the given [`EventHandler`] with common
698 /// functionality implemented by other handlers.
699 /// * [`P2PGossipSync`] if given will update the [`NetworkGraph`] based on payment failures.
701 /// # Rapid Gossip Sync
703 /// If rapid gossip sync is meant to run at startup, pass [`RapidGossipSync`] via `gossip_sync`
704 /// to indicate that the [`BackgroundProcessor`] should not prune the [`NetworkGraph`] instance
705 /// until the [`RapidGossipSync`] instance completes its first sync.
707 /// [top-level documentation]: BackgroundProcessor
708 /// [`join`]: Self::join
709 /// [`stop`]: Self::stop
710 /// [`ChannelManager`]: lightning::ln::channelmanager::ChannelManager
711 /// [`ChannelManager::write`]: lightning::ln::channelmanager::ChannelManager#impl-Writeable
712 /// [`Persister::persist_manager`]: lightning::util::persist::Persister::persist_manager
713 /// [`Persister::persist_graph`]: lightning::util::persist::Persister::persist_graph
714 /// [`NetworkGraph`]: lightning::routing::gossip::NetworkGraph
715 /// [`NetworkGraph::write`]: lightning::routing::gossip::NetworkGraph#impl-Writeable
718 UL: 'static + Deref + Send + Sync,
719 CF: 'static + Deref + Send + Sync,
720 CW: 'static + Deref + Send + Sync,
721 T: 'static + Deref + Send + Sync,
722 ES: 'static + Deref + Send + Sync,
723 NS: 'static + Deref + Send + Sync,
724 SP: 'static + Deref + Send + Sync,
725 F: 'static + Deref + Send + Sync,
726 R: 'static + Deref + Send + Sync,
727 G: 'static + Deref<Target = NetworkGraph<L>> + Send + Sync,
728 L: 'static + Deref + Send + Sync,
729 P: 'static + Deref + Send + Sync,
730 Descriptor: 'static + SocketDescriptor + Send + Sync,
731 CMH: 'static + Deref + Send + Sync,
732 OMH: 'static + Deref + Send + Sync,
733 RMH: 'static + Deref + Send + Sync,
734 EH: 'static + EventHandler + Send,
735 PS: 'static + Deref + Send,
736 M: 'static + Deref<Target = ChainMonitor<<SP::Target as SignerProvider>::Signer, CF, T, F, L, P>> + Send + Sync,
737 CM: 'static + Deref<Target = ChannelManager<CW, T, ES, NS, SP, F, R, L>> + Send + Sync,
738 PGS: 'static + Deref<Target = P2PGossipSync<G, UL, L>> + Send + Sync,
739 RGS: 'static + Deref<Target = RapidGossipSync<G, L>> + Send,
740 UMH: 'static + Deref + Send + Sync,
741 PM: 'static + Deref<Target = PeerManager<Descriptor, CMH, RMH, OMH, L, UMH, NS>> + Send + Sync,
742 S: 'static + Deref<Target = SC> + Send + Sync,
743 SC: for <'b> WriteableScore<'b>,
745 persister: PS, event_handler: EH, chain_monitor: M, channel_manager: CM,
746 gossip_sync: GossipSync<PGS, RGS, G, UL, L>, peer_manager: PM, logger: L, scorer: Option<S>,
749 UL::Target: 'static + UtxoLookup,
750 CF::Target: 'static + chain::Filter,
751 CW::Target: 'static + chain::Watch<<SP::Target as SignerProvider>::Signer>,
752 T::Target: 'static + BroadcasterInterface,
753 ES::Target: 'static + EntropySource,
754 NS::Target: 'static + NodeSigner,
755 SP::Target: 'static + SignerProvider,
756 F::Target: 'static + FeeEstimator,
757 R::Target: 'static + Router,
758 L::Target: 'static + Logger,
759 P::Target: 'static + Persist<<SP::Target as SignerProvider>::Signer>,
760 CMH::Target: 'static + ChannelMessageHandler,
761 OMH::Target: 'static + OnionMessageHandler,
762 RMH::Target: 'static + RoutingMessageHandler,
763 UMH::Target: 'static + CustomMessageHandler,
764 PS::Target: 'static + Persister<'a, CW, T, ES, NS, SP, F, R, L, SC>,
766 let stop_thread = Arc::new(AtomicBool::new(false));
767 let stop_thread_clone = stop_thread.clone();
768 let handle = thread::spawn(move || -> Result<(), std::io::Error> {
769 let event_handler = |event| {
770 let network_graph = gossip_sync.network_graph();
771 if let Some(network_graph) = network_graph {
772 handle_network_graph_update(network_graph, &event)
774 if let Some(ref scorer) = scorer {
775 update_scorer(scorer, &event);
777 event_handler.handle_event(event);
779 define_run_body!(persister, chain_monitor, chain_monitor.process_pending_events(&event_handler),
780 channel_manager, channel_manager.process_pending_events(&event_handler),
781 gossip_sync, peer_manager, logger, scorer, stop_thread.load(Ordering::Acquire),
782 Sleeper::from_two_futures(
783 channel_manager.get_persistable_update_future(),
784 chain_monitor.get_update_future()
785 ).wait_timeout(Duration::from_millis(100)),
786 |_| Instant::now(), |time: &Instant, dur| time.elapsed().as_secs() > dur, false)
788 Self { stop_thread: stop_thread_clone, thread_handle: Some(handle) }
791 /// Join `BackgroundProcessor`'s thread, returning any error that occurred while persisting
792 /// [`ChannelManager`].
796 /// This function panics if the background thread has panicked such as while persisting or
799 /// [`ChannelManager`]: lightning::ln::channelmanager::ChannelManager
800 pub fn join(mut self) -> Result<(), std::io::Error> {
801 assert!(self.thread_handle.is_some());
805 /// Stop `BackgroundProcessor`'s thread, returning any error that occurred while persisting
806 /// [`ChannelManager`].
810 /// This function panics if the background thread has panicked such as while persisting or
813 /// [`ChannelManager`]: lightning::ln::channelmanager::ChannelManager
814 pub fn stop(mut self) -> Result<(), std::io::Error> {
815 assert!(self.thread_handle.is_some());
816 self.stop_and_join_thread()
819 fn stop_and_join_thread(&mut self) -> Result<(), std::io::Error> {
820 self.stop_thread.store(true, Ordering::Release);
824 fn join_thread(&mut self) -> Result<(), std::io::Error> {
825 match self.thread_handle.take() {
826 Some(handle) => handle.join().unwrap(),
832 #[cfg(feature = "std")]
833 impl Drop for BackgroundProcessor {
835 self.stop_and_join_thread().unwrap();
839 #[cfg(all(feature = "std", test))]
841 use bitcoin::blockdata::block::BlockHeader;
842 use bitcoin::blockdata::constants::genesis_block;
843 use bitcoin::blockdata::locktime::PackedLockTime;
844 use bitcoin::blockdata::transaction::{Transaction, TxOut};
845 use bitcoin::network::constants::Network;
846 use bitcoin::secp256k1::{SecretKey, PublicKey, Secp256k1};
847 use lightning::chain::{BestBlock, Confirm, chainmonitor};
848 use lightning::chain::channelmonitor::ANTI_REORG_DELAY;
849 use lightning::chain::keysinterface::{InMemorySigner, KeysManager};
850 use lightning::chain::transaction::OutPoint;
851 use lightning::events::{Event, PathFailure, MessageSendEventsProvider, MessageSendEvent};
852 use lightning::{get_event_msg, get_event};
853 use lightning::ln::PaymentHash;
854 use lightning::ln::channelmanager;
855 use lightning::ln::channelmanager::{BREAKDOWN_TIMEOUT, ChainParameters, MIN_CLTV_EXPIRY_DELTA, PaymentId};
856 use lightning::ln::features::{ChannelFeatures, NodeFeatures};
857 use lightning::ln::msgs::{ChannelMessageHandler, Init};
858 use lightning::ln::peer_handler::{PeerManager, MessageHandler, SocketDescriptor, IgnoringMessageHandler};
859 use lightning::routing::gossip::{NetworkGraph, NodeId, P2PGossipSync};
860 use lightning::routing::router::{DefaultRouter, Path, RouteHop};
861 use lightning::routing::scoring::{ChannelUsage, Score};
862 use lightning::util::config::UserConfig;
863 use lightning::util::ser::Writeable;
864 use lightning::util::test_utils;
865 use lightning::util::persist::KVStorePersister;
866 use lightning_persister::FilesystemPersister;
867 use std::collections::VecDeque;
869 use std::path::PathBuf;
870 use std::sync::{Arc, Mutex};
871 use std::sync::mpsc::SyncSender;
872 use std::time::Duration;
873 use bitcoin::hashes::Hash;
874 use bitcoin::TxMerkleNode;
875 use lightning_rapid_gossip_sync::RapidGossipSync;
876 use super::{BackgroundProcessor, GossipSync, FRESHNESS_TIMER};
878 const EVENT_DEADLINE: u64 = 5 * FRESHNESS_TIMER;
880 #[derive(Clone, Hash, PartialEq, Eq)]
881 struct TestDescriptor{}
882 impl SocketDescriptor for TestDescriptor {
883 fn send_data(&mut self, _data: &[u8], _resume_read: bool) -> usize {
887 fn disconnect_socket(&mut self) {}
890 type ChannelManager = channelmanager::ChannelManager<Arc<ChainMonitor>, Arc<test_utils::TestBroadcaster>, Arc<KeysManager>, Arc<KeysManager>, Arc<KeysManager>, Arc<test_utils::TestFeeEstimator>, Arc<DefaultRouter< Arc<NetworkGraph<Arc<test_utils::TestLogger>>>, Arc<test_utils::TestLogger>, Arc<Mutex<TestScorer>>>>, Arc<test_utils::TestLogger>>;
892 type ChainMonitor = chainmonitor::ChainMonitor<InMemorySigner, Arc<test_utils::TestChainSource>, Arc<test_utils::TestBroadcaster>, Arc<test_utils::TestFeeEstimator>, Arc<test_utils::TestLogger>, Arc<FilesystemPersister>>;
894 type PGS = Arc<P2PGossipSync<Arc<NetworkGraph<Arc<test_utils::TestLogger>>>, Arc<test_utils::TestChainSource>, Arc<test_utils::TestLogger>>>;
895 type RGS = Arc<RapidGossipSync<Arc<NetworkGraph<Arc<test_utils::TestLogger>>>, Arc<test_utils::TestLogger>>>;
898 node: Arc<ChannelManager>,
899 p2p_gossip_sync: PGS,
900 rapid_gossip_sync: RGS,
901 peer_manager: Arc<PeerManager<TestDescriptor, Arc<test_utils::TestChannelMessageHandler>, Arc<test_utils::TestRoutingMessageHandler>, IgnoringMessageHandler, Arc<test_utils::TestLogger>, IgnoringMessageHandler, Arc<KeysManager>>>,
902 chain_monitor: Arc<ChainMonitor>,
903 persister: Arc<FilesystemPersister>,
904 tx_broadcaster: Arc<test_utils::TestBroadcaster>,
905 network_graph: Arc<NetworkGraph<Arc<test_utils::TestLogger>>>,
906 logger: Arc<test_utils::TestLogger>,
907 best_block: BestBlock,
908 scorer: Arc<Mutex<TestScorer>>,
912 fn p2p_gossip_sync(&self) -> GossipSync<PGS, RGS, Arc<NetworkGraph<Arc<test_utils::TestLogger>>>, Arc<test_utils::TestChainSource>, Arc<test_utils::TestLogger>> {
913 GossipSync::P2P(self.p2p_gossip_sync.clone())
916 fn rapid_gossip_sync(&self) -> GossipSync<PGS, RGS, Arc<NetworkGraph<Arc<test_utils::TestLogger>>>, Arc<test_utils::TestChainSource>, Arc<test_utils::TestLogger>> {
917 GossipSync::Rapid(self.rapid_gossip_sync.clone())
920 fn no_gossip_sync(&self) -> GossipSync<PGS, RGS, Arc<NetworkGraph<Arc<test_utils::TestLogger>>>, Arc<test_utils::TestChainSource>, Arc<test_utils::TestLogger>> {
927 let data_dir = self.persister.get_data_dir();
928 match fs::remove_dir_all(data_dir.clone()) {
929 Err(e) => println!("Failed to remove test persister directory {}: {}", data_dir, e),
936 graph_error: Option<(std::io::ErrorKind, &'static str)>,
937 graph_persistence_notifier: Option<SyncSender<()>>,
938 manager_error: Option<(std::io::ErrorKind, &'static str)>,
939 scorer_error: Option<(std::io::ErrorKind, &'static str)>,
940 filesystem_persister: FilesystemPersister,
944 fn new(data_dir: String) -> Self {
945 let filesystem_persister = FilesystemPersister::new(data_dir);
946 Self { graph_error: None, graph_persistence_notifier: None, manager_error: None, scorer_error: None, filesystem_persister }
949 fn with_graph_error(self, error: std::io::ErrorKind, message: &'static str) -> Self {
950 Self { graph_error: Some((error, message)), ..self }
953 fn with_graph_persistence_notifier(self, sender: SyncSender<()>) -> Self {
954 Self { graph_persistence_notifier: Some(sender), ..self }
957 fn with_manager_error(self, error: std::io::ErrorKind, message: &'static str) -> Self {
958 Self { manager_error: Some((error, message)), ..self }
961 fn with_scorer_error(self, error: std::io::ErrorKind, message: &'static str) -> Self {
962 Self { scorer_error: Some((error, message)), ..self }
966 impl KVStorePersister for Persister {
967 fn persist<W: Writeable>(&self, key: &str, object: &W) -> std::io::Result<()> {
968 if key == "manager" {
969 if let Some((error, message)) = self.manager_error {
970 return Err(std::io::Error::new(error, message))
974 if key == "network_graph" {
975 if let Some(sender) = &self.graph_persistence_notifier {
976 match sender.send(()) {
978 Err(std::sync::mpsc::SendError(())) => println!("Persister failed to notify as receiver went away."),
982 if let Some((error, message)) = self.graph_error {
983 return Err(std::io::Error::new(error, message))
988 if let Some((error, message)) = self.scorer_error {
989 return Err(std::io::Error::new(error, message))
993 self.filesystem_persister.persist(key, object)
998 event_expectations: Option<VecDeque<TestResult>>,
1003 PaymentFailure { path: Path, short_channel_id: u64 },
1004 PaymentSuccess { path: Path },
1005 ProbeFailure { path: Path },
1006 ProbeSuccess { path: Path },
1011 Self { event_expectations: None }
1014 fn expect(&mut self, expectation: TestResult) {
1015 self.event_expectations.get_or_insert_with(VecDeque::new).push_back(expectation);
1019 impl lightning::util::ser::Writeable for TestScorer {
1020 fn write<W: lightning::util::ser::Writer>(&self, _: &mut W) -> Result<(), lightning::io::Error> { Ok(()) }
1023 impl Score for TestScorer {
1024 fn channel_penalty_msat(
1025 &self, _short_channel_id: u64, _source: &NodeId, _target: &NodeId, _usage: ChannelUsage
1026 ) -> u64 { unimplemented!(); }
1028 fn payment_path_failed(&mut self, actual_path: &Path, actual_short_channel_id: u64) {
1029 if let Some(expectations) = &mut self.event_expectations {
1030 match expectations.pop_front().unwrap() {
1031 TestResult::PaymentFailure { path, short_channel_id } => {
1032 assert_eq!(actual_path, &path);
1033 assert_eq!(actual_short_channel_id, short_channel_id);
1035 TestResult::PaymentSuccess { path } => {
1036 panic!("Unexpected successful payment path: {:?}", path)
1038 TestResult::ProbeFailure { path } => {
1039 panic!("Unexpected probe failure: {:?}", path)
1041 TestResult::ProbeSuccess { path } => {
1042 panic!("Unexpected probe success: {:?}", path)
1048 fn payment_path_successful(&mut self, actual_path: &Path) {
1049 if let Some(expectations) = &mut self.event_expectations {
1050 match expectations.pop_front().unwrap() {
1051 TestResult::PaymentFailure { path, .. } => {
1052 panic!("Unexpected payment path failure: {:?}", path)
1054 TestResult::PaymentSuccess { path } => {
1055 assert_eq!(actual_path, &path);
1057 TestResult::ProbeFailure { path } => {
1058 panic!("Unexpected probe failure: {:?}", path)
1060 TestResult::ProbeSuccess { path } => {
1061 panic!("Unexpected probe success: {:?}", path)
1067 fn probe_failed(&mut self, actual_path: &Path, _: u64) {
1068 if let Some(expectations) = &mut self.event_expectations {
1069 match expectations.pop_front().unwrap() {
1070 TestResult::PaymentFailure { path, .. } => {
1071 panic!("Unexpected payment path failure: {:?}", path)
1073 TestResult::PaymentSuccess { path } => {
1074 panic!("Unexpected payment path success: {:?}", path)
1076 TestResult::ProbeFailure { path } => {
1077 assert_eq!(actual_path, &path);
1079 TestResult::ProbeSuccess { path } => {
1080 panic!("Unexpected probe success: {:?}", path)
1085 fn probe_successful(&mut self, actual_path: &Path) {
1086 if let Some(expectations) = &mut self.event_expectations {
1087 match expectations.pop_front().unwrap() {
1088 TestResult::PaymentFailure { path, .. } => {
1089 panic!("Unexpected payment path failure: {:?}", path)
1091 TestResult::PaymentSuccess { path } => {
1092 panic!("Unexpected payment path success: {:?}", path)
1094 TestResult::ProbeFailure { path } => {
1095 panic!("Unexpected probe failure: {:?}", path)
1097 TestResult::ProbeSuccess { path } => {
1098 assert_eq!(actual_path, &path);
1105 impl Drop for TestScorer {
1106 fn drop(&mut self) {
1107 if std::thread::panicking() {
1111 if let Some(event_expectations) = &self.event_expectations {
1112 if !event_expectations.is_empty() {
1113 panic!("Unsatisfied event expectations: {:?}", event_expectations);
1119 fn get_full_filepath(filepath: String, filename: String) -> String {
1120 let mut path = PathBuf::from(filepath);
1121 path.push(filename);
1122 path.to_str().unwrap().to_string()
1125 fn create_nodes(num_nodes: usize, persist_dir: String) -> Vec<Node> {
1126 let network = Network::Testnet;
1127 let mut nodes = Vec::new();
1128 for i in 0..num_nodes {
1129 let tx_broadcaster = Arc::new(test_utils::TestBroadcaster::new(network));
1130 let fee_estimator = Arc::new(test_utils::TestFeeEstimator { sat_per_kw: Mutex::new(253) });
1131 let logger = Arc::new(test_utils::TestLogger::with_id(format!("node {}", i)));
1132 let genesis_block = genesis_block(network);
1133 let network_graph = Arc::new(NetworkGraph::new(network, logger.clone()));
1134 let scorer = Arc::new(Mutex::new(TestScorer::new()));
1135 let seed = [i as u8; 32];
1136 let router = Arc::new(DefaultRouter::new(network_graph.clone(), logger.clone(), seed, scorer.clone()));
1137 let chain_source = Arc::new(test_utils::TestChainSource::new(Network::Testnet));
1138 let persister = Arc::new(FilesystemPersister::new(format!("{}_persister_{}", persist_dir, i)));
1139 let now = Duration::from_secs(genesis_block.header.time as u64);
1140 let keys_manager = Arc::new(KeysManager::new(&seed, now.as_secs(), now.subsec_nanos()));
1141 let chain_monitor = Arc::new(chainmonitor::ChainMonitor::new(Some(chain_source.clone()), tx_broadcaster.clone(), logger.clone(), fee_estimator.clone(), persister.clone()));
1142 let best_block = BestBlock::from_network(network);
1143 let params = ChainParameters { network, best_block };
1144 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));
1145 let p2p_gossip_sync = Arc::new(P2PGossipSync::new(network_graph.clone(), Some(chain_source.clone()), logger.clone()));
1146 let rapid_gossip_sync = Arc::new(RapidGossipSync::new(network_graph.clone(), logger.clone()));
1147 let msg_handler = MessageHandler { chan_handler: Arc::new(test_utils::TestChannelMessageHandler::new()), route_handler: Arc::new(test_utils::TestRoutingMessageHandler::new()), onion_message_handler: IgnoringMessageHandler{}};
1148 let peer_manager = Arc::new(PeerManager::new(msg_handler, 0, &seed, logger.clone(), IgnoringMessageHandler{}, keys_manager.clone()));
1149 let node = Node { node: manager, p2p_gossip_sync, rapid_gossip_sync, peer_manager, chain_monitor, persister, tx_broadcaster, network_graph, logger, best_block, scorer };
1153 for i in 0..num_nodes {
1154 for j in (i+1)..num_nodes {
1155 nodes[i].node.peer_connected(&nodes[j].node.get_our_node_id(), &Init { features: nodes[j].node.init_features(), remote_network_address: None }, true).unwrap();
1156 nodes[j].node.peer_connected(&nodes[i].node.get_our_node_id(), &Init { features: nodes[i].node.init_features(), remote_network_address: None }, false).unwrap();
1163 macro_rules! open_channel {
1164 ($node_a: expr, $node_b: expr, $channel_value: expr) => {{
1165 begin_open_channel!($node_a, $node_b, $channel_value);
1166 let events = $node_a.node.get_and_clear_pending_events();
1167 assert_eq!(events.len(), 1);
1168 let (temporary_channel_id, tx) = handle_funding_generation_ready!(events[0], $channel_value);
1169 $node_a.node.funding_transaction_generated(&temporary_channel_id, &$node_b.node.get_our_node_id(), tx.clone()).unwrap();
1170 $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()));
1171 get_event!($node_b, Event::ChannelPending);
1172 $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()));
1173 get_event!($node_a, Event::ChannelPending);
1178 macro_rules! begin_open_channel {
1179 ($node_a: expr, $node_b: expr, $channel_value: expr) => {{
1180 $node_a.node.create_channel($node_b.node.get_our_node_id(), $channel_value, 100, 42, None).unwrap();
1181 $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()));
1182 $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()));
1186 macro_rules! handle_funding_generation_ready {
1187 ($event: expr, $channel_value: expr) => {{
1189 Event::FundingGenerationReady { temporary_channel_id, channel_value_satoshis, ref output_script, user_channel_id, .. } => {
1190 assert_eq!(channel_value_satoshis, $channel_value);
1191 assert_eq!(user_channel_id, 42);
1193 let tx = Transaction { version: 1 as i32, lock_time: PackedLockTime(0), input: Vec::new(), output: vec![TxOut {
1194 value: channel_value_satoshis, script_pubkey: output_script.clone(),
1196 (temporary_channel_id, tx)
1198 _ => panic!("Unexpected event"),
1203 fn confirm_transaction_depth(node: &mut Node, tx: &Transaction, depth: u32) {
1204 for i in 1..=depth {
1205 let prev_blockhash = node.best_block.block_hash();
1206 let height = node.best_block.height() + 1;
1207 let header = BlockHeader { version: 0x20000000, prev_blockhash, merkle_root: TxMerkleNode::all_zeros(), time: height, bits: 42, nonce: 42 };
1208 let txdata = vec![(0, tx)];
1209 node.best_block = BestBlock::new(header.block_hash(), height);
1212 node.node.transactions_confirmed(&header, &txdata, height);
1213 node.chain_monitor.transactions_confirmed(&header, &txdata, height);
1215 x if x == depth => {
1216 node.node.best_block_updated(&header, height);
1217 node.chain_monitor.best_block_updated(&header, height);
1223 fn confirm_transaction(node: &mut Node, tx: &Transaction) {
1224 confirm_transaction_depth(node, tx, ANTI_REORG_DELAY);
1228 fn test_background_processor() {
1229 // Test that when a new channel is created, the ChannelManager needs to be re-persisted with
1230 // updates. Also test that when new updates are available, the manager signals that it needs
1231 // re-persistence and is successfully re-persisted.
1232 let nodes = create_nodes(2, "test_background_processor".to_string());
1234 // Go through the channel creation process so that each node has something to persist. Since
1235 // open_channel consumes events, it must complete before starting BackgroundProcessor to
1236 // avoid a race with processing events.
1237 let tx = open_channel!(nodes[0], nodes[1], 100000);
1239 // Initiate the background processors to watch each node.
1240 let data_dir = nodes[0].persister.get_data_dir();
1241 let persister = Arc::new(Persister::new(data_dir));
1242 let event_handler = |_: _| {};
1243 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()));
1245 macro_rules! check_persisted_data {
1246 ($node: expr, $filepath: expr) => {
1247 let mut expected_bytes = Vec::new();
1249 expected_bytes.clear();
1250 match $node.write(&mut expected_bytes) {
1252 match std::fs::read($filepath) {
1254 if bytes == expected_bytes {
1263 Err(e) => panic!("Unexpected error: {}", e)
1269 // Check that the initial channel manager data is persisted as expected.
1270 let filepath = get_full_filepath("test_background_processor_persister_0".to_string(), "manager".to_string());
1271 check_persisted_data!(nodes[0].node, filepath.clone());
1274 if !nodes[0].node.get_persistence_condvar_value() { break }
1277 // Force-close the channel.
1278 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();
1280 // Check that the force-close updates are persisted.
1281 check_persisted_data!(nodes[0].node, filepath.clone());
1283 if !nodes[0].node.get_persistence_condvar_value() { break }
1286 // Check network graph is persisted
1287 let filepath = get_full_filepath("test_background_processor_persister_0".to_string(), "network_graph".to_string());
1288 check_persisted_data!(nodes[0].network_graph, filepath.clone());
1290 // Check scorer is persisted
1291 let filepath = get_full_filepath("test_background_processor_persister_0".to_string(), "scorer".to_string());
1292 check_persisted_data!(nodes[0].scorer, filepath.clone());
1294 if !std::thread::panicking() {
1295 bg_processor.stop().unwrap();
1300 fn test_timer_tick_called() {
1301 // Test that `ChannelManager::timer_tick_occurred` is called every `FRESHNESS_TIMER`,
1302 // `ChainMonitor::rebroadcast_pending_claims` is called every `REBROADCAST_TIMER`, and
1303 // `PeerManager::timer_tick_occurred` every `PING_TIMER`.
1304 let nodes = create_nodes(1, "test_timer_tick_called".to_string());
1305 let data_dir = nodes[0].persister.get_data_dir();
1306 let persister = Arc::new(Persister::new(data_dir));
1307 let event_handler = |_: _| {};
1308 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()));
1310 let log_entries = nodes[0].logger.lines.lock().unwrap();
1311 let desired_log_1 = "Calling ChannelManager's timer_tick_occurred".to_string();
1312 let desired_log_2 = "Calling PeerManager's timer_tick_occurred".to_string();
1313 let desired_log_3 = "Rebroadcasting monitor's pending claims".to_string();
1314 if log_entries.get(&("lightning_background_processor".to_string(), desired_log_1)).is_some() &&
1315 log_entries.get(&("lightning_background_processor".to_string(), desired_log_2)).is_some() &&
1316 log_entries.get(&("lightning_background_processor".to_string(), desired_log_3)).is_some() {
1321 if !std::thread::panicking() {
1322 bg_processor.stop().unwrap();
1327 fn test_channel_manager_persist_error() {
1328 // Test that if we encounter an error during manager persistence, the thread panics.
1329 let nodes = create_nodes(2, "test_persist_error".to_string());
1330 open_channel!(nodes[0], nodes[1], 100000);
1332 let data_dir = nodes[0].persister.get_data_dir();
1333 let persister = Arc::new(Persister::new(data_dir).with_manager_error(std::io::ErrorKind::Other, "test"));
1334 let event_handler = |_: _| {};
1335 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()));
1336 match bg_processor.join() {
1337 Ok(_) => panic!("Expected error persisting manager"),
1339 assert_eq!(e.kind(), std::io::ErrorKind::Other);
1340 assert_eq!(e.get_ref().unwrap().to_string(), "test");
1346 #[cfg(feature = "futures")]
1347 async fn test_channel_manager_persist_error_async() {
1348 // Test that if we encounter an error during manager persistence, the thread panics.
1349 let nodes = create_nodes(2, "test_persist_error_sync".to_string());
1350 open_channel!(nodes[0], nodes[1], 100000);
1352 let data_dir = nodes[0].persister.get_data_dir();
1353 let persister = Arc::new(Persister::new(data_dir).with_manager_error(std::io::ErrorKind::Other, "test"));
1355 let bp_future = super::process_events_async(
1356 persister, |_: _| {async {}}, nodes[0].chain_monitor.clone(), nodes[0].node.clone(),
1357 nodes[0].rapid_gossip_sync(), nodes[0].peer_manager.clone(), nodes[0].logger.clone(),
1358 Some(nodes[0].scorer.clone()), move |dur: Duration| {
1359 Box::pin(async move {
1360 tokio::time::sleep(dur).await;
1365 match bp_future.await {
1366 Ok(_) => panic!("Expected error persisting manager"),
1368 assert_eq!(e.kind(), std::io::ErrorKind::Other);
1369 assert_eq!(e.get_ref().unwrap().to_string(), "test");
1375 fn test_network_graph_persist_error() {
1376 // Test that if we encounter an error during network graph persistence, an error gets returned.
1377 let nodes = create_nodes(2, "test_persist_network_graph_error".to_string());
1378 let data_dir = nodes[0].persister.get_data_dir();
1379 let persister = Arc::new(Persister::new(data_dir).with_graph_error(std::io::ErrorKind::Other, "test"));
1380 let event_handler = |_: _| {};
1381 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()));
1383 match bg_processor.stop() {
1384 Ok(_) => panic!("Expected error persisting network graph"),
1386 assert_eq!(e.kind(), std::io::ErrorKind::Other);
1387 assert_eq!(e.get_ref().unwrap().to_string(), "test");
1393 fn test_scorer_persist_error() {
1394 // Test that if we encounter an error during scorer persistence, an error gets returned.
1395 let nodes = create_nodes(2, "test_persist_scorer_error".to_string());
1396 let data_dir = nodes[0].persister.get_data_dir();
1397 let persister = Arc::new(Persister::new(data_dir).with_scorer_error(std::io::ErrorKind::Other, "test"));
1398 let event_handler = |_: _| {};
1399 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()));
1401 match bg_processor.stop() {
1402 Ok(_) => panic!("Expected error persisting scorer"),
1404 assert_eq!(e.kind(), std::io::ErrorKind::Other);
1405 assert_eq!(e.get_ref().unwrap().to_string(), "test");
1411 fn test_background_event_handling() {
1412 let mut nodes = create_nodes(2, "test_background_event_handling".to_string());
1413 let channel_value = 100000;
1414 let data_dir = nodes[0].persister.get_data_dir();
1415 let persister = Arc::new(Persister::new(data_dir.clone()));
1417 // Set up a background event handler for FundingGenerationReady events.
1418 let (funding_generation_send, funding_generation_recv) = std::sync::mpsc::sync_channel(1);
1419 let (channel_pending_send, channel_pending_recv) = std::sync::mpsc::sync_channel(1);
1420 let event_handler = move |event: Event| match event {
1421 Event::FundingGenerationReady { .. } => funding_generation_send.send(handle_funding_generation_ready!(event, channel_value)).unwrap(),
1422 Event::ChannelPending { .. } => channel_pending_send.send(()).unwrap(),
1423 Event::ChannelReady { .. } => {},
1424 _ => panic!("Unexpected event: {:?}", event),
1427 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()));
1429 // Open a channel and check that the FundingGenerationReady event was handled.
1430 begin_open_channel!(nodes[0], nodes[1], channel_value);
1431 let (temporary_channel_id, funding_tx) = funding_generation_recv
1432 .recv_timeout(Duration::from_secs(EVENT_DEADLINE))
1433 .expect("FundingGenerationReady not handled within deadline");
1434 nodes[0].node.funding_transaction_generated(&temporary_channel_id, &nodes[1].node.get_our_node_id(), funding_tx.clone()).unwrap();
1435 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()));
1436 get_event!(nodes[1], Event::ChannelPending);
1437 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()));
1438 let _ = channel_pending_recv.recv_timeout(Duration::from_secs(EVENT_DEADLINE))
1439 .expect("ChannelPending not handled within deadline");
1441 // Confirm the funding transaction.
1442 confirm_transaction(&mut nodes[0], &funding_tx);
1443 let as_funding = get_event_msg!(nodes[0], MessageSendEvent::SendChannelReady, nodes[1].node.get_our_node_id());
1444 confirm_transaction(&mut nodes[1], &funding_tx);
1445 let bs_funding = get_event_msg!(nodes[1], MessageSendEvent::SendChannelReady, nodes[0].node.get_our_node_id());
1446 nodes[0].node.handle_channel_ready(&nodes[1].node.get_our_node_id(), &bs_funding);
1447 let _as_channel_update = get_event_msg!(nodes[0], MessageSendEvent::SendChannelUpdate, nodes[1].node.get_our_node_id());
1448 nodes[1].node.handle_channel_ready(&nodes[0].node.get_our_node_id(), &as_funding);
1449 let _bs_channel_update = get_event_msg!(nodes[1], MessageSendEvent::SendChannelUpdate, nodes[0].node.get_our_node_id());
1451 if !std::thread::panicking() {
1452 bg_processor.stop().unwrap();
1455 // Set up a background event handler for SpendableOutputs events.
1456 let (sender, receiver) = std::sync::mpsc::sync_channel(1);
1457 let event_handler = move |event: Event| match event {
1458 Event::SpendableOutputs { .. } => sender.send(event).unwrap(),
1459 Event::ChannelReady { .. } => {},
1460 Event::ChannelClosed { .. } => {},
1461 _ => panic!("Unexpected event: {:?}", event),
1463 let persister = Arc::new(Persister::new(data_dir));
1464 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()));
1466 // Force close the channel and check that the SpendableOutputs event was handled.
1467 nodes[0].node.force_close_broadcasting_latest_txn(&nodes[0].node.list_channels()[0].channel_id, &nodes[1].node.get_our_node_id()).unwrap();
1468 let commitment_tx = nodes[0].tx_broadcaster.txn_broadcasted.lock().unwrap().pop().unwrap();
1469 confirm_transaction_depth(&mut nodes[0], &commitment_tx, BREAKDOWN_TIMEOUT as u32);
1471 let event = receiver
1472 .recv_timeout(Duration::from_secs(EVENT_DEADLINE))
1473 .expect("Events not handled within deadline");
1475 Event::SpendableOutputs { .. } => {},
1476 _ => panic!("Unexpected event: {:?}", event),
1479 if !std::thread::panicking() {
1480 bg_processor.stop().unwrap();
1485 fn test_scorer_persistence() {
1486 let nodes = create_nodes(2, "test_scorer_persistence".to_string());
1487 let data_dir = nodes[0].persister.get_data_dir();
1488 let persister = Arc::new(Persister::new(data_dir));
1489 let event_handler = |_: _| {};
1490 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()));
1493 let log_entries = nodes[0].logger.lines.lock().unwrap();
1494 let expected_log = "Persisting scorer".to_string();
1495 if log_entries.get(&("lightning_background_processor".to_string(), expected_log)).is_some() {
1500 if !std::thread::panicking() {
1501 bg_processor.stop().unwrap();
1505 macro_rules! do_test_not_pruning_network_graph_until_graph_sync_completion {
1506 ($nodes: expr, $receive: expr, $sleep: expr) => {
1507 let features = ChannelFeatures::empty();
1508 $nodes[0].network_graph.add_channel_from_partial_announcement(
1509 42, 53, features, $nodes[0].node.get_our_node_id(), $nodes[1].node.get_our_node_id()
1510 ).expect("Failed to update channel from partial announcement");
1511 let original_graph_description = $nodes[0].network_graph.to_string();
1512 assert!(original_graph_description.contains("42: features: 0000, node_one:"));
1513 assert_eq!($nodes[0].network_graph.read_only().channels().len(), 1);
1517 let log_entries = $nodes[0].logger.lines.lock().unwrap();
1518 let loop_counter = "Calling ChannelManager's timer_tick_occurred".to_string();
1519 if *log_entries.get(&("lightning_background_processor".to_string(), loop_counter))
1522 // Wait until the loop has gone around at least twice.
1527 let initialization_input = vec![
1528 76, 68, 75, 1, 111, 226, 140, 10, 182, 241, 179, 114, 193, 166, 162, 70, 174, 99, 247,
1529 79, 147, 30, 131, 101, 225, 90, 8, 156, 104, 214, 25, 0, 0, 0, 0, 0, 97, 227, 98, 218,
1530 0, 0, 0, 4, 2, 22, 7, 207, 206, 25, 164, 197, 231, 230, 231, 56, 102, 61, 250, 251,
1531 187, 172, 38, 46, 79, 247, 108, 44, 155, 48, 219, 238, 252, 53, 192, 6, 67, 2, 36, 125,
1532 157, 176, 223, 175, 234, 116, 94, 248, 201, 225, 97, 235, 50, 47, 115, 172, 63, 136,
1533 88, 216, 115, 11, 111, 217, 114, 84, 116, 124, 231, 107, 2, 158, 1, 242, 121, 152, 106,
1534 204, 131, 186, 35, 93, 70, 216, 10, 237, 224, 183, 89, 95, 65, 3, 83, 185, 58, 138,
1535 181, 64, 187, 103, 127, 68, 50, 2, 201, 19, 17, 138, 136, 149, 185, 226, 156, 137, 175,
1536 110, 32, 237, 0, 217, 90, 31, 100, 228, 149, 46, 219, 175, 168, 77, 4, 143, 38, 128,
1537 76, 97, 0, 0, 0, 2, 0, 0, 255, 8, 153, 192, 0, 2, 27, 0, 0, 0, 1, 0, 0, 255, 2, 68,
1538 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,
1539 0, 0, 0, 1, 0, 0, 0, 0, 58, 85, 116, 216, 255, 8, 153, 192, 0, 2, 27, 0, 0, 25, 0, 0,
1540 0, 1, 0, 0, 0, 125, 255, 2, 68, 226, 0, 6, 11, 0, 1, 5, 0, 0, 0, 0, 29, 129, 25, 192,
1542 $nodes[0].rapid_gossip_sync.update_network_graph_no_std(&initialization_input[..], Some(1642291930)).unwrap();
1544 // this should have added two channels and pruned the previous one.
1545 assert_eq!($nodes[0].network_graph.read_only().channels().len(), 2);
1547 $receive.expect("Network graph not pruned within deadline");
1549 // all channels should now be pruned
1550 assert_eq!($nodes[0].network_graph.read_only().channels().len(), 0);
1555 fn test_not_pruning_network_graph_until_graph_sync_completion() {
1556 let (sender, receiver) = std::sync::mpsc::sync_channel(1);
1558 let nodes = create_nodes(2, "test_not_pruning_network_graph_until_graph_sync_completion".to_string());
1559 let data_dir = nodes[0].persister.get_data_dir();
1560 let persister = Arc::new(Persister::new(data_dir).with_graph_persistence_notifier(sender));
1562 let event_handler = |_: _| {};
1563 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()));
1565 do_test_not_pruning_network_graph_until_graph_sync_completion!(nodes,
1566 receiver.recv_timeout(Duration::from_secs(super::FIRST_NETWORK_PRUNE_TIMER * 5)),
1567 std::thread::sleep(Duration::from_millis(1)));
1569 background_processor.stop().unwrap();
1573 #[cfg(feature = "futures")]
1574 async fn test_not_pruning_network_graph_until_graph_sync_completion_async() {
1575 let (sender, receiver) = std::sync::mpsc::sync_channel(1);
1577 let nodes = create_nodes(2, "test_not_pruning_network_graph_until_graph_sync_completion_async".to_string());
1578 let data_dir = nodes[0].persister.get_data_dir();
1579 let persister = Arc::new(Persister::new(data_dir).with_graph_persistence_notifier(sender));
1581 let (exit_sender, exit_receiver) = tokio::sync::watch::channel(());
1582 let bp_future = super::process_events_async(
1583 persister, |_: _| {async {}}, nodes[0].chain_monitor.clone(), nodes[0].node.clone(),
1584 nodes[0].rapid_gossip_sync(), nodes[0].peer_manager.clone(), nodes[0].logger.clone(),
1585 Some(nodes[0].scorer.clone()), move |dur: Duration| {
1586 let mut exit_receiver = exit_receiver.clone();
1587 Box::pin(async move {
1589 _ = tokio::time::sleep(dur) => false,
1590 _ = exit_receiver.changed() => true,
1596 let t1 = tokio::spawn(bp_future);
1597 let t2 = tokio::spawn(async move {
1598 do_test_not_pruning_network_graph_until_graph_sync_completion!(nodes, {
1601 tokio::time::sleep(Duration::from_secs(super::FIRST_NETWORK_PRUNE_TIMER)).await;
1602 if let Ok(()) = receiver.try_recv() { break Ok::<(), ()>(()); }
1606 }, tokio::time::sleep(Duration::from_millis(1)).await);
1607 exit_sender.send(()).unwrap();
1609 let (r1, r2) = tokio::join!(t1, t2);
1610 r1.unwrap().unwrap();
1614 macro_rules! do_test_payment_path_scoring {
1615 ($nodes: expr, $receive: expr) => {
1616 // Ensure that we update the scorer when relevant events are processed. In this case, we ensure
1617 // that we update the scorer upon a payment path succeeding (note that the channel must be
1618 // public or else we won't score it).
1619 // A background event handler for FundingGenerationReady events must be hooked up to a
1620 // running background processor.
1621 let scored_scid = 4242;
1622 let secp_ctx = Secp256k1::new();
1623 let node_1_privkey = SecretKey::from_slice(&[42; 32]).unwrap();
1624 let node_1_id = PublicKey::from_secret_key(&secp_ctx, &node_1_privkey);
1626 let path = Path { hops: vec![RouteHop {
1628 node_features: NodeFeatures::empty(),
1629 short_channel_id: scored_scid,
1630 channel_features: ChannelFeatures::empty(),
1632 cltv_expiry_delta: MIN_CLTV_EXPIRY_DELTA as u32,
1633 }], blinded_tail: None };
1635 $nodes[0].scorer.lock().unwrap().expect(TestResult::PaymentFailure { path: path.clone(), short_channel_id: scored_scid });
1636 $nodes[0].node.push_pending_event(Event::PaymentPathFailed {
1638 payment_hash: PaymentHash([42; 32]),
1639 payment_failed_permanently: false,
1640 failure: PathFailure::OnPath { network_update: None },
1642 short_channel_id: Some(scored_scid),
1644 let event = $receive.expect("PaymentPathFailed not handled within deadline");
1646 Event::PaymentPathFailed { .. } => {},
1647 _ => panic!("Unexpected event"),
1650 // Ensure we'll score payments that were explicitly failed back by the destination as
1652 $nodes[0].scorer.lock().unwrap().expect(TestResult::ProbeSuccess { path: path.clone() });
1653 $nodes[0].node.push_pending_event(Event::PaymentPathFailed {
1655 payment_hash: PaymentHash([42; 32]),
1656 payment_failed_permanently: true,
1657 failure: PathFailure::OnPath { network_update: None },
1659 short_channel_id: None,
1661 let event = $receive.expect("PaymentPathFailed not handled within deadline");
1663 Event::PaymentPathFailed { .. } => {},
1664 _ => panic!("Unexpected event"),
1667 $nodes[0].scorer.lock().unwrap().expect(TestResult::PaymentSuccess { path: path.clone() });
1668 $nodes[0].node.push_pending_event(Event::PaymentPathSuccessful {
1669 payment_id: PaymentId([42; 32]),
1673 let event = $receive.expect("PaymentPathSuccessful not handled within deadline");
1675 Event::PaymentPathSuccessful { .. } => {},
1676 _ => panic!("Unexpected event"),
1679 $nodes[0].scorer.lock().unwrap().expect(TestResult::ProbeSuccess { path: path.clone() });
1680 $nodes[0].node.push_pending_event(Event::ProbeSuccessful {
1681 payment_id: PaymentId([42; 32]),
1682 payment_hash: PaymentHash([42; 32]),
1685 let event = $receive.expect("ProbeSuccessful not handled within deadline");
1687 Event::ProbeSuccessful { .. } => {},
1688 _ => panic!("Unexpected event"),
1691 $nodes[0].scorer.lock().unwrap().expect(TestResult::ProbeFailure { path: path.clone() });
1692 $nodes[0].node.push_pending_event(Event::ProbeFailed {
1693 payment_id: PaymentId([42; 32]),
1694 payment_hash: PaymentHash([42; 32]),
1696 short_channel_id: Some(scored_scid),
1698 let event = $receive.expect("ProbeFailure not handled within deadline");
1700 Event::ProbeFailed { .. } => {},
1701 _ => panic!("Unexpected event"),
1707 fn test_payment_path_scoring() {
1708 let (sender, receiver) = std::sync::mpsc::sync_channel(1);
1709 let event_handler = move |event: Event| match event {
1710 Event::PaymentPathFailed { .. } => sender.send(event).unwrap(),
1711 Event::PaymentPathSuccessful { .. } => sender.send(event).unwrap(),
1712 Event::ProbeSuccessful { .. } => sender.send(event).unwrap(),
1713 Event::ProbeFailed { .. } => sender.send(event).unwrap(),
1714 _ => panic!("Unexpected event: {:?}", event),
1717 let nodes = create_nodes(1, "test_payment_path_scoring".to_string());
1718 let data_dir = nodes[0].persister.get_data_dir();
1719 let persister = Arc::new(Persister::new(data_dir));
1720 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()));
1722 do_test_payment_path_scoring!(nodes, receiver.recv_timeout(Duration::from_secs(EVENT_DEADLINE)));
1724 if !std::thread::panicking() {
1725 bg_processor.stop().unwrap();
1730 #[cfg(feature = "futures")]
1731 async fn test_payment_path_scoring_async() {
1732 let (sender, mut receiver) = tokio::sync::mpsc::channel(1);
1733 let event_handler = move |event: Event| {
1734 let sender_ref = sender.clone();
1737 Event::PaymentPathFailed { .. } => { sender_ref.send(event).await.unwrap() },
1738 Event::PaymentPathSuccessful { .. } => { sender_ref.send(event).await.unwrap() },
1739 Event::ProbeSuccessful { .. } => { sender_ref.send(event).await.unwrap() },
1740 Event::ProbeFailed { .. } => { sender_ref.send(event).await.unwrap() },
1741 _ => panic!("Unexpected event: {:?}", event),
1746 let nodes = create_nodes(1, "test_payment_path_scoring_async".to_string());
1747 let data_dir = nodes[0].persister.get_data_dir();
1748 let persister = Arc::new(Persister::new(data_dir));
1750 let (exit_sender, exit_receiver) = tokio::sync::watch::channel(());
1752 let bp_future = super::process_events_async(
1753 persister, event_handler, nodes[0].chain_monitor.clone(), nodes[0].node.clone(),
1754 nodes[0].no_gossip_sync(), nodes[0].peer_manager.clone(), nodes[0].logger.clone(),
1755 Some(nodes[0].scorer.clone()), move |dur: Duration| {
1756 let mut exit_receiver = exit_receiver.clone();
1757 Box::pin(async move {
1759 _ = tokio::time::sleep(dur) => false,
1760 _ = exit_receiver.changed() => true,
1765 let t1 = tokio::spawn(bp_future);
1766 let t2 = tokio::spawn(async move {
1767 do_test_payment_path_scoring!(nodes, receiver.recv().await);
1768 exit_sender.send(()).unwrap();
1771 let (r1, r2) = tokio::join!(t1, t2);
1772 r1.unwrap().unwrap();