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::{Score, WriteableScore};
38 use lightning::util::logger::Logger;
39 use lightning::util::persist::Persister;
40 #[cfg(feature = "std")]
41 use lightning::util::wakers::Sleeper;
42 use lightning_rapid_gossip_sync::RapidGossipSync;
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
244 let mut score = scorer.lock();
246 Event::PaymentPathFailed { ref path, short_channel_id: Some(scid), .. } => {
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 score.probe_successful(path);
254 Event::PaymentPathSuccessful { path, .. } => {
255 score.payment_path_successful(path);
257 Event::ProbeSuccessful { path, .. } => {
258 score.probe_successful(path);
260 Event::ProbeFailed { path, short_channel_id: Some(scid), .. } => {
261 score.probe_failed(path, *scid);
268 macro_rules! define_run_body {
269 ($persister: ident, $chain_monitor: ident, $process_chain_monitor_events: expr,
270 $channel_manager: ident, $process_channel_manager_events: expr,
271 $gossip_sync: ident, $peer_manager: ident, $logger: ident, $scorer: ident,
272 $loop_exit_check: expr, $await: expr, $get_timer: expr, $timer_elapsed: expr,
273 $check_slow_await: expr)
275 log_trace!($logger, "Calling ChannelManager's timer_tick_occurred on startup");
276 $channel_manager.timer_tick_occurred();
277 log_trace!($logger, "Rebroadcasting monitor's pending claims on startup");
278 $chain_monitor.rebroadcast_pending_claims();
280 let mut last_freshness_call = $get_timer(FRESHNESS_TIMER);
281 let mut last_ping_call = $get_timer(PING_TIMER);
282 let mut last_prune_call = $get_timer(FIRST_NETWORK_PRUNE_TIMER);
283 let mut last_scorer_persist_call = $get_timer(SCORER_PERSIST_TIMER);
284 let mut last_rebroadcast_call = $get_timer(REBROADCAST_TIMER);
285 let mut have_pruned = false;
288 $process_channel_manager_events;
289 $process_chain_monitor_events;
291 // Note that the PeerManager::process_events may block on ChannelManager's locks,
292 // hence it comes last here. When the ChannelManager finishes whatever it's doing,
293 // we want to ensure we get into `persist_manager` as quickly as we can, especially
294 // without running the normal event processing above and handing events to users.
296 // Specifically, on an *extremely* slow machine, we may see ChannelManager start
297 // processing a message effectively at any point during this loop. In order to
298 // minimize the time between such processing completing and persisting the updated
299 // ChannelManager, we want to minimize methods blocking on a ChannelManager
300 // generally, and as a fallback place such blocking only immediately before
302 $peer_manager.as_ref().process_events();
304 // Exit the loop if the background processor was requested to stop.
305 if $loop_exit_check {
306 log_trace!($logger, "Terminating background processor.");
310 // We wait up to 100ms, but track how long it takes to detect being put to sleep,
311 // see `await_start`'s use below.
312 let mut await_start = None;
313 if $check_slow_await { await_start = Some($get_timer(1)); }
314 let updates_available = $await;
315 let await_slow = if $check_slow_await { $timer_elapsed(&mut await_start.unwrap(), 1) } else { false };
317 // Exit the loop if the background processor was requested to stop.
318 if $loop_exit_check {
319 log_trace!($logger, "Terminating background processor.");
323 if updates_available {
324 log_trace!($logger, "Persisting ChannelManager...");
325 $persister.persist_manager(&*$channel_manager)?;
326 log_trace!($logger, "Done persisting ChannelManager.");
328 if $timer_elapsed(&mut last_freshness_call, FRESHNESS_TIMER) {
329 log_trace!($logger, "Calling ChannelManager's timer_tick_occurred");
330 $channel_manager.timer_tick_occurred();
331 last_freshness_call = $get_timer(FRESHNESS_TIMER);
334 // On various platforms, we may be starved of CPU cycles for several reasons.
335 // E.g. on iOS, if we've been in the background, we will be entirely paused.
336 // Similarly, if we're on a desktop platform and the device has been asleep, we
337 // may not get any cycles.
338 // We detect this by checking if our max-100ms-sleep, above, ran longer than a
339 // full second, at which point we assume sockets may have been killed (they
340 // appear to be at least on some platforms, even if it has only been a second).
341 // Note that we have to take care to not get here just because user event
342 // processing was slow at the top of the loop. For example, the sample client
343 // may call Bitcoin Core RPCs during event handling, which very often takes
344 // more than a handful of seconds to complete, and shouldn't disconnect all our
346 log_trace!($logger, "100ms sleep took more than a second, disconnecting peers.");
347 $peer_manager.as_ref().disconnect_all_peers();
348 last_ping_call = $get_timer(PING_TIMER);
349 } else if $timer_elapsed(&mut last_ping_call, PING_TIMER) {
350 log_trace!($logger, "Calling PeerManager's timer_tick_occurred");
351 $peer_manager.as_ref().timer_tick_occurred();
352 last_ping_call = $get_timer(PING_TIMER);
355 // Note that we want to run a graph prune once not long after startup before
356 // falling back to our usual hourly prunes. This avoids short-lived clients never
357 // pruning their network graph. We run once 60 seconds after startup before
358 // continuing our normal cadence. For RGS, since 60 seconds is likely too long,
359 // we prune after an initial sync completes.
360 let prune_timer = if have_pruned { NETWORK_PRUNE_TIMER } else { FIRST_NETWORK_PRUNE_TIMER };
361 let prune_timer_elapsed = $timer_elapsed(&mut last_prune_call, prune_timer);
362 let should_prune = match $gossip_sync {
363 GossipSync::Rapid(_) => !have_pruned || prune_timer_elapsed,
364 _ => prune_timer_elapsed,
367 // The network graph must not be pruned while rapid sync completion is pending
368 if let Some(network_graph) = $gossip_sync.prunable_network_graph() {
369 #[cfg(feature = "std")] {
370 log_trace!($logger, "Pruning and persisting network graph.");
371 network_graph.remove_stale_channels_and_tracking();
373 #[cfg(not(feature = "std"))] {
374 log_warn!($logger, "Not pruning network graph, consider enabling `std` or doing so manually with remove_stale_channels_and_tracking_with_time.");
375 log_trace!($logger, "Persisting network graph.");
378 if let Err(e) = $persister.persist_graph(network_graph) {
379 log_error!($logger, "Error: Failed to persist network graph, check your disk and permissions {}", e)
384 let prune_timer = if have_pruned { NETWORK_PRUNE_TIMER } else { FIRST_NETWORK_PRUNE_TIMER };
385 last_prune_call = $get_timer(prune_timer);
388 if $timer_elapsed(&mut last_scorer_persist_call, SCORER_PERSIST_TIMER) {
389 if let Some(ref scorer) = $scorer {
390 log_trace!($logger, "Persisting scorer");
391 if let Err(e) = $persister.persist_scorer(&scorer) {
392 log_error!($logger, "Error: Failed to persist scorer, check your disk and permissions {}", e)
395 last_scorer_persist_call = $get_timer(SCORER_PERSIST_TIMER);
398 if $timer_elapsed(&mut last_rebroadcast_call, REBROADCAST_TIMER) {
399 log_trace!($logger, "Rebroadcasting monitor's pending claims");
400 $chain_monitor.rebroadcast_pending_claims();
401 last_rebroadcast_call = $get_timer(REBROADCAST_TIMER);
405 // After we exit, ensure we persist the ChannelManager one final time - this avoids
406 // some races where users quit while channel updates were in-flight, with
407 // ChannelMonitor update(s) persisted without a corresponding ChannelManager update.
408 $persister.persist_manager(&*$channel_manager)?;
410 // Persist Scorer on exit
411 if let Some(ref scorer) = $scorer {
412 $persister.persist_scorer(&scorer)?;
415 // Persist NetworkGraph on exit
416 if let Some(network_graph) = $gossip_sync.network_graph() {
417 $persister.persist_graph(network_graph)?;
424 #[cfg(feature = "futures")]
425 pub(crate) mod futures_util {
426 use core::future::Future;
427 use core::task::{Poll, Waker, RawWaker, RawWakerVTable};
429 use core::marker::Unpin;
430 pub(crate) struct Selector<
431 A: Future<Output=()> + Unpin, B: Future<Output=()> + Unpin, C: Future<Output=bool> + Unpin
437 pub(crate) enum SelectorOutput {
442 A: Future<Output=()> + Unpin, B: Future<Output=()> + Unpin, C: Future<Output=bool> + Unpin
443 > Future for Selector<A, B, C> {
444 type Output = SelectorOutput;
445 fn poll(mut self: Pin<&mut Self>, ctx: &mut core::task::Context<'_>) -> Poll<SelectorOutput> {
446 match Pin::new(&mut self.a).poll(ctx) {
447 Poll::Ready(()) => { return Poll::Ready(SelectorOutput::A); },
450 match Pin::new(&mut self.b).poll(ctx) {
451 Poll::Ready(()) => { return Poll::Ready(SelectorOutput::B); },
454 match Pin::new(&mut self.c).poll(ctx) {
455 Poll::Ready(res) => { return Poll::Ready(SelectorOutput::C(res)); },
462 // If we want to poll a future without an async context to figure out if it has completed or
463 // not without awaiting, we need a Waker, which needs a vtable...we fill it with dummy values
464 // but sadly there's a good bit of boilerplate here.
465 fn dummy_waker_clone(_: *const ()) -> RawWaker { RawWaker::new(core::ptr::null(), &DUMMY_WAKER_VTABLE) }
466 fn dummy_waker_action(_: *const ()) { }
468 const DUMMY_WAKER_VTABLE: RawWakerVTable = RawWakerVTable::new(
469 dummy_waker_clone, dummy_waker_action, dummy_waker_action, dummy_waker_action);
470 pub(crate) fn dummy_waker() -> Waker { unsafe { Waker::from_raw(RawWaker::new(core::ptr::null(), &DUMMY_WAKER_VTABLE)) } }
472 #[cfg(feature = "futures")]
473 use futures_util::{Selector, SelectorOutput, dummy_waker};
474 #[cfg(feature = "futures")]
477 /// Processes background events in a future.
479 /// `sleeper` should return a future which completes in the given amount of time and returns a
480 /// boolean indicating whether the background processing should exit. Once `sleeper` returns a
481 /// future which outputs `true`, the loop will exit and this function's future will complete.
482 /// The `sleeper` future is free to return early after it has triggered the exit condition.
484 /// See [`BackgroundProcessor::start`] for information on which actions this handles.
486 /// Requires the `futures` feature. Note that while this method is available without the `std`
487 /// feature, doing so will skip calling [`NetworkGraph::remove_stale_channels_and_tracking`],
488 /// you should call [`NetworkGraph::remove_stale_channels_and_tracking_with_time`] regularly
489 /// manually instead.
491 /// The `mobile_interruptable_platform` flag should be set if we're currently running on a
492 /// mobile device, where we may need to check for interruption of the application regularly. If you
493 /// are unsure, you should set the flag, as the performance impact of it is minimal unless there
494 /// are hundreds or thousands of simultaneous process calls running.
496 /// For example, in order to process background events in a [Tokio](https://tokio.rs/) task, you
497 /// could setup `process_events_async` like this:
499 /// # struct MyPersister {}
500 /// # impl lightning::util::persist::KVStorePersister for MyPersister {
501 /// # fn persist<W: lightning::util::ser::Writeable>(&self, key: &str, object: &W) -> lightning::io::Result<()> { Ok(()) }
503 /// # struct MyEventHandler {}
504 /// # impl MyEventHandler {
505 /// # async fn handle_event(&self, _: lightning::events::Event) {}
507 /// # #[derive(Eq, PartialEq, Clone, Hash)]
508 /// # struct MySocketDescriptor {}
509 /// # impl lightning::ln::peer_handler::SocketDescriptor for MySocketDescriptor {
510 /// # fn send_data(&mut self, _data: &[u8], _resume_read: bool) -> usize { 0 }
511 /// # fn disconnect_socket(&mut self) {}
513 /// # use std::sync::{Arc, Mutex};
514 /// # use std::sync::atomic::{AtomicBool, Ordering};
515 /// # use lightning_background_processor::{process_events_async, GossipSync};
516 /// # type MyBroadcaster = dyn lightning::chain::chaininterface::BroadcasterInterface + Send + Sync;
517 /// # type MyFeeEstimator = dyn lightning::chain::chaininterface::FeeEstimator + Send + Sync;
518 /// # type MyNodeSigner = dyn lightning::sign::NodeSigner + Send + Sync;
519 /// # type MyUtxoLookup = dyn lightning::routing::utxo::UtxoLookup + Send + Sync;
520 /// # type MyFilter = dyn lightning::chain::Filter + Send + Sync;
521 /// # type MyLogger = dyn lightning::util::logger::Logger + Send + Sync;
522 /// # type MyMessageRouter = dyn lightning::onion_message::MessageRouter + Send + Sync;
523 /// # type MyChainMonitor = lightning::chain::chainmonitor::ChainMonitor<lightning::sign::InMemorySigner, Arc<MyFilter>, Arc<MyBroadcaster>, Arc<MyFeeEstimator>, Arc<MyLogger>, Arc<MyPersister>>;
524 /// # type MyPeerManager = lightning::ln::peer_handler::SimpleArcPeerManager<MySocketDescriptor, MyChainMonitor, MyBroadcaster, MyFeeEstimator, MyUtxoLookup, MyLogger, MyMessageRouter>;
525 /// # type MyNetworkGraph = lightning::routing::gossip::NetworkGraph<Arc<MyLogger>>;
526 /// # type MyGossipSync = lightning::routing::gossip::P2PGossipSync<Arc<MyNetworkGraph>, Arc<MyUtxoLookup>, Arc<MyLogger>>;
527 /// # type MyChannelManager = lightning::ln::channelmanager::SimpleArcChannelManager<MyChainMonitor, MyBroadcaster, MyFeeEstimator, MyLogger>;
528 /// # type MyScorer = Mutex<lightning::routing::scoring::ProbabilisticScorer<Arc<MyNetworkGraph>, Arc<MyLogger>>>;
530 /// # 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>) {
531 /// let background_persister = Arc::clone(&my_persister);
532 /// let background_event_handler = Arc::clone(&my_event_handler);
533 /// let background_chain_mon = Arc::clone(&my_chain_monitor);
534 /// let background_chan_man = Arc::clone(&my_channel_manager);
535 /// let background_gossip_sync = GossipSync::p2p(Arc::clone(&my_gossip_sync));
536 /// let background_peer_man = Arc::clone(&my_peer_manager);
537 /// let background_logger = Arc::clone(&my_logger);
538 /// let background_scorer = Arc::clone(&my_scorer);
540 /// // Setup the sleeper.
541 /// let (stop_sender, stop_receiver) = tokio::sync::watch::channel(());
543 /// let sleeper = move |d| {
544 /// let mut receiver = stop_receiver.clone();
545 /// Box::pin(async move {
547 /// _ = tokio::time::sleep(d) => false,
548 /// _ = receiver.changed() => true,
553 /// let mobile_interruptable_platform = false;
555 /// let handle = tokio::spawn(async move {
556 /// process_events_async(
557 /// background_persister,
558 /// |e| background_event_handler.handle_event(e),
559 /// background_chain_mon,
560 /// background_chan_man,
561 /// background_gossip_sync,
562 /// background_peer_man,
563 /// background_logger,
564 /// Some(background_scorer),
566 /// mobile_interruptable_platform,
569 /// .expect("Failed to process events");
572 /// // Stop the background processing.
573 /// stop_sender.send(()).unwrap();
574 /// handle.await.unwrap();
577 #[cfg(feature = "futures")]
578 pub async fn process_events_async<
580 UL: 'static + Deref + Send + Sync,
581 CF: 'static + Deref + Send + Sync,
582 CW: 'static + Deref + Send + Sync,
583 T: 'static + Deref + Send + Sync,
584 ES: 'static + Deref + Send + Sync,
585 NS: 'static + Deref + Send + Sync,
586 SP: 'static + Deref + Send + Sync,
587 F: 'static + Deref + Send + Sync,
588 R: 'static + Deref + Send + Sync,
589 G: 'static + Deref<Target = NetworkGraph<L>> + Send + Sync,
590 L: 'static + Deref + Send + Sync,
591 P: 'static + Deref + Send + Sync,
592 EventHandlerFuture: core::future::Future<Output = ()>,
593 EventHandler: Fn(Event) -> EventHandlerFuture,
594 PS: 'static + Deref + Send,
595 M: 'static + Deref<Target = ChainMonitor<<SP::Target as SignerProvider>::Signer, CF, T, F, L, P>> + Send + Sync,
596 CM: 'static + Deref<Target = ChannelManager<CW, T, ES, NS, SP, F, R, L>> + Send + Sync,
597 PGS: 'static + Deref<Target = P2PGossipSync<G, UL, L>> + Send + Sync,
598 RGS: 'static + Deref<Target = RapidGossipSync<G, L>> + Send,
599 APM: APeerManager + Send + Sync,
600 PM: 'static + Deref<Target = APM> + Send + Sync,
601 S: 'static + Deref<Target = SC> + Send + Sync,
602 SC: for<'b> WriteableScore<'b>,
603 SleepFuture: core::future::Future<Output = bool> + core::marker::Unpin,
604 Sleeper: Fn(Duration) -> SleepFuture
606 persister: PS, event_handler: EventHandler, chain_monitor: M, channel_manager: CM,
607 gossip_sync: GossipSync<PGS, RGS, G, UL, L>, peer_manager: PM, logger: L, scorer: Option<S>,
608 sleeper: Sleeper, mobile_interruptable_platform: bool,
609 ) -> Result<(), lightning::io::Error>
611 UL::Target: 'static + UtxoLookup,
612 CF::Target: 'static + chain::Filter,
613 CW::Target: 'static + chain::Watch<<SP::Target as SignerProvider>::Signer>,
614 T::Target: 'static + BroadcasterInterface,
615 ES::Target: 'static + EntropySource,
616 NS::Target: 'static + NodeSigner,
617 SP::Target: 'static + SignerProvider,
618 F::Target: 'static + FeeEstimator,
619 R::Target: 'static + Router,
620 L::Target: 'static + Logger,
621 P::Target: 'static + Persist<<SP::Target as SignerProvider>::Signer>,
622 PS::Target: 'static + Persister<'a, CW, T, ES, NS, SP, F, R, L, SC>,
624 let mut should_break = false;
625 let async_event_handler = |event| {
626 let network_graph = gossip_sync.network_graph();
627 let event_handler = &event_handler;
628 let scorer = &scorer;
629 let logger = &logger;
630 let persister = &persister;
632 if let Some(network_graph) = network_graph {
633 handle_network_graph_update(network_graph, &event)
635 if let Some(ref scorer) = scorer {
636 if update_scorer(scorer, &event) {
637 log_trace!(logger, "Persisting scorer after update");
638 if let Err(e) = persister.persist_scorer(&scorer) {
639 log_error!(logger, "Error: Failed to persist scorer, check your disk and permissions {}", e)
643 event_handler(event).await;
646 define_run_body!(persister,
647 chain_monitor, chain_monitor.process_pending_events_async(async_event_handler).await,
648 channel_manager, channel_manager.process_pending_events_async(async_event_handler).await,
649 gossip_sync, peer_manager, logger, scorer, should_break, {
651 a: channel_manager.get_persistable_update_future(),
652 b: chain_monitor.get_update_future(),
653 c: sleeper(if mobile_interruptable_platform { Duration::from_millis(100) } else { Duration::from_secs(FASTEST_TIMER) }),
656 SelectorOutput::A => true,
657 SelectorOutput::B => false,
658 SelectorOutput::C(exit) => {
663 }, |t| sleeper(Duration::from_secs(t)),
664 |fut: &mut SleepFuture, _| {
665 let mut waker = dummy_waker();
666 let mut ctx = task::Context::from_waker(&mut waker);
667 match core::pin::Pin::new(fut).poll(&mut ctx) {
668 task::Poll::Ready(exit) => { should_break = exit; true },
669 task::Poll::Pending => false,
671 }, mobile_interruptable_platform)
674 #[cfg(feature = "std")]
675 impl BackgroundProcessor {
676 /// Start a background thread that takes care of responsibilities enumerated in the [top-level
679 /// The thread runs indefinitely unless the object is dropped, [`stop`] is called, or
680 /// [`Persister::persist_manager`] returns an error. In case of an error, the error is retrieved by calling
681 /// either [`join`] or [`stop`].
683 /// # Data Persistence
685 /// [`Persister::persist_manager`] is responsible for writing out the [`ChannelManager`] to disk, and/or
686 /// uploading to one or more backup services. See [`ChannelManager::write`] for writing out a
687 /// [`ChannelManager`]. See the `lightning-persister` crate for LDK's
688 /// provided implementation.
690 /// [`Persister::persist_graph`] is responsible for writing out the [`NetworkGraph`] to disk, if
691 /// [`GossipSync`] is supplied. See [`NetworkGraph::write`] for writing out a [`NetworkGraph`].
692 /// See the `lightning-persister` crate for LDK's provided implementation.
694 /// Typically, users should either implement [`Persister::persist_manager`] to never return an
695 /// error or call [`join`] and handle any error that may arise. For the latter case,
696 /// `BackgroundProcessor` must be restarted by calling `start` again after handling the error.
700 /// `event_handler` is responsible for handling events that users should be notified of (e.g.,
701 /// payment failed). [`BackgroundProcessor`] may decorate the given [`EventHandler`] with common
702 /// functionality implemented by other handlers.
703 /// * [`P2PGossipSync`] if given will update the [`NetworkGraph`] based on payment failures.
705 /// # Rapid Gossip Sync
707 /// If rapid gossip sync is meant to run at startup, pass [`RapidGossipSync`] via `gossip_sync`
708 /// to indicate that the [`BackgroundProcessor`] should not prune the [`NetworkGraph`] instance
709 /// until the [`RapidGossipSync`] instance completes its first sync.
711 /// [top-level documentation]: BackgroundProcessor
712 /// [`join`]: Self::join
713 /// [`stop`]: Self::stop
714 /// [`ChannelManager`]: lightning::ln::channelmanager::ChannelManager
715 /// [`ChannelManager::write`]: lightning::ln::channelmanager::ChannelManager#impl-Writeable
716 /// [`Persister::persist_manager`]: lightning::util::persist::Persister::persist_manager
717 /// [`Persister::persist_graph`]: lightning::util::persist::Persister::persist_graph
718 /// [`NetworkGraph`]: lightning::routing::gossip::NetworkGraph
719 /// [`NetworkGraph::write`]: lightning::routing::gossip::NetworkGraph#impl-Writeable
722 UL: 'static + Deref + Send + Sync,
723 CF: 'static + Deref + Send + Sync,
724 CW: 'static + Deref + Send + Sync,
725 T: 'static + Deref + Send + Sync,
726 ES: 'static + Deref + Send + Sync,
727 NS: 'static + Deref + Send + Sync,
728 SP: 'static + Deref + Send + Sync,
729 F: 'static + Deref + Send + Sync,
730 R: 'static + Deref + Send + Sync,
731 G: 'static + Deref<Target = NetworkGraph<L>> + Send + Sync,
732 L: 'static + Deref + Send + Sync,
733 P: '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 APM: APeerManager + Send + Sync,
741 PM: 'static + Deref<Target = APM> + 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 PS::Target: 'static + Persister<'a, CW, T, ES, NS, SP, F, R, L, SC>,
762 let stop_thread = Arc::new(AtomicBool::new(false));
763 let stop_thread_clone = stop_thread.clone();
764 let handle = thread::spawn(move || -> Result<(), std::io::Error> {
765 let event_handler = |event| {
766 let network_graph = gossip_sync.network_graph();
767 if let Some(network_graph) = network_graph {
768 handle_network_graph_update(network_graph, &event)
770 if let Some(ref scorer) = scorer {
771 if update_scorer(scorer, &event) {
772 log_trace!(logger, "Persisting scorer after update");
773 if let Err(e) = persister.persist_scorer(&scorer) {
774 log_error!(logger, "Error: Failed to persist scorer, check your disk and permissions {}", e)
778 event_handler.handle_event(event);
780 define_run_body!(persister, chain_monitor, chain_monitor.process_pending_events(&event_handler),
781 channel_manager, channel_manager.process_pending_events(&event_handler),
782 gossip_sync, peer_manager, logger, scorer, stop_thread.load(Ordering::Acquire),
783 Sleeper::from_two_futures(
784 channel_manager.get_persistable_update_future(),
785 chain_monitor.get_update_future()
786 ).wait_timeout(Duration::from_millis(100)),
787 |_| Instant::now(), |time: &Instant, dur| time.elapsed().as_secs() > dur, false)
789 Self { stop_thread: stop_thread_clone, thread_handle: Some(handle) }
792 /// Join `BackgroundProcessor`'s thread, returning any error that occurred while persisting
793 /// [`ChannelManager`].
797 /// This function panics if the background thread has panicked such as while persisting or
800 /// [`ChannelManager`]: lightning::ln::channelmanager::ChannelManager
801 pub fn join(mut self) -> Result<(), std::io::Error> {
802 assert!(self.thread_handle.is_some());
806 /// Stop `BackgroundProcessor`'s thread, returning any error that occurred while persisting
807 /// [`ChannelManager`].
811 /// This function panics if the background thread has panicked such as while persisting or
814 /// [`ChannelManager`]: lightning::ln::channelmanager::ChannelManager
815 pub fn stop(mut self) -> Result<(), std::io::Error> {
816 assert!(self.thread_handle.is_some());
817 self.stop_and_join_thread()
820 fn stop_and_join_thread(&mut self) -> Result<(), std::io::Error> {
821 self.stop_thread.store(true, Ordering::Release);
825 fn join_thread(&mut self) -> Result<(), std::io::Error> {
826 match self.thread_handle.take() {
827 Some(handle) => handle.join().unwrap(),
833 #[cfg(feature = "std")]
834 impl Drop for BackgroundProcessor {
836 self.stop_and_join_thread().unwrap();
840 #[cfg(all(feature = "std", test))]
842 use bitcoin::blockdata::constants::{genesis_block, ChainHash};
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::sign::{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::functional_test_utils::*;
858 use lightning::ln::msgs::{ChannelMessageHandler, Init};
859 use lightning::ln::peer_handler::{PeerManager, MessageHandler, SocketDescriptor, IgnoringMessageHandler};
860 use lightning::routing::gossip::{NetworkGraph, NodeId, P2PGossipSync};
861 use lightning::routing::router::{DefaultRouter, Path, RouteHop};
862 use lightning::routing::scoring::{ChannelUsage, Score};
863 use lightning::util::config::UserConfig;
864 use lightning::util::ser::Writeable;
865 use lightning::util::test_utils;
866 use lightning::util::persist::KVStorePersister;
867 use lightning_persister::FilesystemPersister;
868 use std::collections::VecDeque;
870 use std::path::PathBuf;
871 use std::sync::{Arc, Mutex};
872 use std::sync::mpsc::SyncSender;
873 use std::time::Duration;
874 use lightning_rapid_gossip_sync::RapidGossipSync;
875 use super::{BackgroundProcessor, GossipSync, FRESHNESS_TIMER};
877 const EVENT_DEADLINE: u64 = 5 * FRESHNESS_TIMER;
879 #[derive(Clone, Hash, PartialEq, Eq)]
880 struct TestDescriptor{}
881 impl SocketDescriptor for TestDescriptor {
882 fn send_data(&mut self, _data: &[u8], _resume_read: bool) -> usize {
886 fn disconnect_socket(&mut self) {}
889 type ChannelManager = channelmanager::ChannelManager<Arc<ChainMonitor>, Arc<test_utils::TestBroadcaster>, Arc<KeysManager>, Arc<KeysManager>, Arc<KeysManager>, Arc<test_utils::TestFeeEstimator>, Arc<DefaultRouter<Arc<NetworkGraph<Arc<test_utils::TestLogger>>>, Arc<test_utils::TestLogger>, Arc<Mutex<TestScorer>>, (), TestScorer>>, Arc<test_utils::TestLogger>>;
891 type ChainMonitor = chainmonitor::ChainMonitor<InMemorySigner, Arc<test_utils::TestChainSource>, Arc<test_utils::TestBroadcaster>, Arc<test_utils::TestFeeEstimator>, Arc<test_utils::TestLogger>, Arc<FilesystemPersister>>;
893 type PGS = Arc<P2PGossipSync<Arc<NetworkGraph<Arc<test_utils::TestLogger>>>, Arc<test_utils::TestChainSource>, Arc<test_utils::TestLogger>>>;
894 type RGS = Arc<RapidGossipSync<Arc<NetworkGraph<Arc<test_utils::TestLogger>>>, Arc<test_utils::TestLogger>>>;
897 node: Arc<ChannelManager>,
898 p2p_gossip_sync: PGS,
899 rapid_gossip_sync: RGS,
900 peer_manager: Arc<PeerManager<TestDescriptor, Arc<test_utils::TestChannelMessageHandler>, Arc<test_utils::TestRoutingMessageHandler>, IgnoringMessageHandler, Arc<test_utils::TestLogger>, IgnoringMessageHandler, Arc<KeysManager>>>,
901 chain_monitor: Arc<ChainMonitor>,
902 persister: Arc<FilesystemPersister>,
903 tx_broadcaster: Arc<test_utils::TestBroadcaster>,
904 network_graph: Arc<NetworkGraph<Arc<test_utils::TestLogger>>>,
905 logger: Arc<test_utils::TestLogger>,
906 best_block: BestBlock,
907 scorer: Arc<Mutex<TestScorer>>,
911 fn p2p_gossip_sync(&self) -> GossipSync<PGS, RGS, Arc<NetworkGraph<Arc<test_utils::TestLogger>>>, Arc<test_utils::TestChainSource>, Arc<test_utils::TestLogger>> {
912 GossipSync::P2P(self.p2p_gossip_sync.clone())
915 fn rapid_gossip_sync(&self) -> GossipSync<PGS, RGS, Arc<NetworkGraph<Arc<test_utils::TestLogger>>>, Arc<test_utils::TestChainSource>, Arc<test_utils::TestLogger>> {
916 GossipSync::Rapid(self.rapid_gossip_sync.clone())
919 fn no_gossip_sync(&self) -> GossipSync<PGS, RGS, Arc<NetworkGraph<Arc<test_utils::TestLogger>>>, Arc<test_utils::TestChainSource>, Arc<test_utils::TestLogger>> {
926 let data_dir = self.persister.get_data_dir();
927 match fs::remove_dir_all(data_dir.clone()) {
928 Err(e) => println!("Failed to remove test persister directory {}: {}", data_dir, e),
935 graph_error: Option<(std::io::ErrorKind, &'static str)>,
936 graph_persistence_notifier: Option<SyncSender<()>>,
937 manager_error: Option<(std::io::ErrorKind, &'static str)>,
938 scorer_error: Option<(std::io::ErrorKind, &'static str)>,
939 filesystem_persister: FilesystemPersister,
943 fn new(data_dir: String) -> Self {
944 let filesystem_persister = FilesystemPersister::new(data_dir);
945 Self { graph_error: None, graph_persistence_notifier: None, manager_error: None, scorer_error: None, filesystem_persister }
948 fn with_graph_error(self, error: std::io::ErrorKind, message: &'static str) -> Self {
949 Self { graph_error: Some((error, message)), ..self }
952 fn with_graph_persistence_notifier(self, sender: SyncSender<()>) -> Self {
953 Self { graph_persistence_notifier: Some(sender), ..self }
956 fn with_manager_error(self, error: std::io::ErrorKind, message: &'static str) -> Self {
957 Self { manager_error: Some((error, message)), ..self }
960 fn with_scorer_error(self, error: std::io::ErrorKind, message: &'static str) -> Self {
961 Self { scorer_error: Some((error, message)), ..self }
965 impl KVStorePersister for Persister {
966 fn persist<W: Writeable>(&self, key: &str, object: &W) -> std::io::Result<()> {
967 if key == "manager" {
968 if let Some((error, message)) = self.manager_error {
969 return Err(std::io::Error::new(error, message))
973 if key == "network_graph" {
974 if let Some(sender) = &self.graph_persistence_notifier {
975 match sender.send(()) {
977 Err(std::sync::mpsc::SendError(())) => println!("Persister failed to notify as receiver went away."),
981 if let Some((error, message)) = self.graph_error {
982 return Err(std::io::Error::new(error, message))
987 if let Some((error, message)) = self.scorer_error {
988 return Err(std::io::Error::new(error, message))
992 self.filesystem_persister.persist(key, object)
997 event_expectations: Option<VecDeque<TestResult>>,
1002 PaymentFailure { path: Path, short_channel_id: u64 },
1003 PaymentSuccess { path: Path },
1004 ProbeFailure { path: Path },
1005 ProbeSuccess { path: Path },
1010 Self { event_expectations: None }
1013 fn expect(&mut self, expectation: TestResult) {
1014 self.event_expectations.get_or_insert_with(VecDeque::new).push_back(expectation);
1018 impl lightning::util::ser::Writeable for TestScorer {
1019 fn write<W: lightning::util::ser::Writer>(&self, _: &mut W) -> Result<(), lightning::io::Error> { Ok(()) }
1022 impl Score for TestScorer {
1023 type ScoreParams = ();
1024 fn channel_penalty_msat(
1025 &self, _short_channel_id: u64, _source: &NodeId, _target: &NodeId, _usage: ChannelUsage, _score_params: &Self::ScoreParams
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: &str) -> (String, Vec<Node>) {
1126 let persist_temp_path = env::temp_dir().join(persist_dir);
1127 let persist_dir = persist_temp_path.to_string_lossy().to_string();
1128 let network = Network::Bitcoin;
1129 let mut nodes = Vec::new();
1130 for i in 0..num_nodes {
1131 let tx_broadcaster = Arc::new(test_utils::TestBroadcaster::new(network));
1132 let fee_estimator = Arc::new(test_utils::TestFeeEstimator { sat_per_kw: Mutex::new(253) });
1133 let logger = Arc::new(test_utils::TestLogger::with_id(format!("node {}", i)));
1134 let genesis_block = genesis_block(network);
1135 let network_graph = Arc::new(NetworkGraph::new(network, logger.clone()));
1136 let scorer = Arc::new(Mutex::new(TestScorer::new()));
1137 let seed = [i as u8; 32];
1138 let router = Arc::new(DefaultRouter::new(network_graph.clone(), logger.clone(), seed, scorer.clone(), ()));
1139 let chain_source = Arc::new(test_utils::TestChainSource::new(Network::Bitcoin));
1140 let persister = Arc::new(FilesystemPersister::new(format!("{}_persister_{}", &persist_dir, i)));
1141 let now = Duration::from_secs(genesis_block.header.time as u64);
1142 let keys_manager = Arc::new(KeysManager::new(&seed, now.as_secs(), now.subsec_nanos()));
1143 let chain_monitor = Arc::new(chainmonitor::ChainMonitor::new(Some(chain_source.clone()), tx_broadcaster.clone(), logger.clone(), fee_estimator.clone(), persister.clone()));
1144 let best_block = BestBlock::from_network(network);
1145 let params = ChainParameters { network, best_block };
1146 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));
1147 let p2p_gossip_sync = Arc::new(P2PGossipSync::new(network_graph.clone(), Some(chain_source.clone()), logger.clone()));
1148 let rapid_gossip_sync = Arc::new(RapidGossipSync::new(network_graph.clone(), logger.clone()));
1149 let msg_handler = MessageHandler {
1150 chan_handler: Arc::new(test_utils::TestChannelMessageHandler::new(ChainHash::using_genesis_block(Network::Testnet))),
1151 route_handler: Arc::new(test_utils::TestRoutingMessageHandler::new()),
1152 onion_message_handler: IgnoringMessageHandler{}, custom_message_handler: IgnoringMessageHandler{}
1154 let peer_manager = Arc::new(PeerManager::new(msg_handler, 0, &seed, logger.clone(), keys_manager.clone()));
1155 let node = Node { node: manager, p2p_gossip_sync, rapid_gossip_sync, peer_manager, chain_monitor, persister, tx_broadcaster, network_graph, logger, best_block, scorer };
1159 for i in 0..num_nodes {
1160 for j in (i+1)..num_nodes {
1161 nodes[i].node.peer_connected(&nodes[j].node.get_our_node_id(), &Init {
1162 features: nodes[j].node.init_features(), networks: None, remote_network_address: None
1164 nodes[j].node.peer_connected(&nodes[i].node.get_our_node_id(), &Init {
1165 features: nodes[i].node.init_features(), networks: None, remote_network_address: None
1170 (persist_dir, nodes)
1173 macro_rules! open_channel {
1174 ($node_a: expr, $node_b: expr, $channel_value: expr) => {{
1175 begin_open_channel!($node_a, $node_b, $channel_value);
1176 let events = $node_a.node.get_and_clear_pending_events();
1177 assert_eq!(events.len(), 1);
1178 let (temporary_channel_id, tx) = handle_funding_generation_ready!(events[0], $channel_value);
1179 $node_a.node.funding_transaction_generated(&temporary_channel_id, &$node_b.node.get_our_node_id(), tx.clone()).unwrap();
1180 $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()));
1181 get_event!($node_b, Event::ChannelPending);
1182 $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()));
1183 get_event!($node_a, Event::ChannelPending);
1188 macro_rules! begin_open_channel {
1189 ($node_a: expr, $node_b: expr, $channel_value: expr) => {{
1190 $node_a.node.create_channel($node_b.node.get_our_node_id(), $channel_value, 100, 42, None).unwrap();
1191 $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()));
1192 $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()));
1196 macro_rules! handle_funding_generation_ready {
1197 ($event: expr, $channel_value: expr) => {{
1199 Event::FundingGenerationReady { temporary_channel_id, channel_value_satoshis, ref output_script, user_channel_id, .. } => {
1200 assert_eq!(channel_value_satoshis, $channel_value);
1201 assert_eq!(user_channel_id, 42);
1203 let tx = Transaction { version: 1 as i32, lock_time: PackedLockTime(0), input: Vec::new(), output: vec![TxOut {
1204 value: channel_value_satoshis, script_pubkey: output_script.clone(),
1206 (temporary_channel_id, tx)
1208 _ => panic!("Unexpected event"),
1213 fn confirm_transaction_depth(node: &mut Node, tx: &Transaction, depth: u32) {
1214 for i in 1..=depth {
1215 let prev_blockhash = node.best_block.block_hash();
1216 let height = node.best_block.height() + 1;
1217 let header = create_dummy_header(prev_blockhash, height);
1218 let txdata = vec![(0, tx)];
1219 node.best_block = BestBlock::new(header.block_hash(), height);
1222 node.node.transactions_confirmed(&header, &txdata, height);
1223 node.chain_monitor.transactions_confirmed(&header, &txdata, height);
1225 x if x == depth => {
1226 node.node.best_block_updated(&header, height);
1227 node.chain_monitor.best_block_updated(&header, height);
1233 fn confirm_transaction(node: &mut Node, tx: &Transaction) {
1234 confirm_transaction_depth(node, tx, ANTI_REORG_DELAY);
1238 fn test_background_processor() {
1239 // Test that when a new channel is created, the ChannelManager needs to be re-persisted with
1240 // updates. Also test that when new updates are available, the manager signals that it needs
1241 // re-persistence and is successfully re-persisted.
1242 let (persist_dir, nodes) = create_nodes(2, "test_background_processor");
1244 // Go through the channel creation process so that each node has something to persist. Since
1245 // open_channel consumes events, it must complete before starting BackgroundProcessor to
1246 // avoid a race with processing events.
1247 let tx = open_channel!(nodes[0], nodes[1], 100000);
1249 // Initiate the background processors to watch each node.
1250 let data_dir = nodes[0].persister.get_data_dir();
1251 let persister = Arc::new(Persister::new(data_dir));
1252 let event_handler = |_: _| {};
1253 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()));
1255 macro_rules! check_persisted_data {
1256 ($node: expr, $filepath: expr) => {
1257 let mut expected_bytes = Vec::new();
1259 expected_bytes.clear();
1260 match $node.write(&mut expected_bytes) {
1262 match std::fs::read($filepath) {
1264 if bytes == expected_bytes {
1273 Err(e) => panic!("Unexpected error: {}", e)
1279 // Check that the initial channel manager data is persisted as expected.
1280 let filepath = get_full_filepath(format!("{}_persister_0", &persist_dir), "manager".to_string());
1281 check_persisted_data!(nodes[0].node, filepath.clone());
1284 if !nodes[0].node.get_persistence_condvar_value() { break }
1287 // Force-close the channel.
1288 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();
1290 // Check that the force-close updates are persisted.
1291 check_persisted_data!(nodes[0].node, filepath.clone());
1293 if !nodes[0].node.get_persistence_condvar_value() { break }
1296 // Check network graph is persisted
1297 let filepath = get_full_filepath(format!("{}_persister_0", &persist_dir), "network_graph".to_string());
1298 check_persisted_data!(nodes[0].network_graph, filepath.clone());
1300 // Check scorer is persisted
1301 let filepath = get_full_filepath(format!("{}_persister_0", &persist_dir), "scorer".to_string());
1302 check_persisted_data!(nodes[0].scorer, filepath.clone());
1304 if !std::thread::panicking() {
1305 bg_processor.stop().unwrap();
1310 fn test_timer_tick_called() {
1311 // Test that `ChannelManager::timer_tick_occurred` is called every `FRESHNESS_TIMER`,
1312 // `ChainMonitor::rebroadcast_pending_claims` is called every `REBROADCAST_TIMER`, and
1313 // `PeerManager::timer_tick_occurred` every `PING_TIMER`.
1314 let (_, nodes) = create_nodes(1, "test_timer_tick_called");
1315 let data_dir = nodes[0].persister.get_data_dir();
1316 let persister = Arc::new(Persister::new(data_dir));
1317 let event_handler = |_: _| {};
1318 let bg_processor = BackgroundProcessor::start(persister, event_handler, nodes[0].chain_monitor.clone(), nodes[0].node.clone(), nodes[0].no_gossip_sync(), nodes[0].peer_manager.clone(), nodes[0].logger.clone(), Some(nodes[0].scorer.clone()));
1320 let log_entries = nodes[0].logger.lines.lock().unwrap();
1321 let desired_log_1 = "Calling ChannelManager's timer_tick_occurred".to_string();
1322 let desired_log_2 = "Calling PeerManager's timer_tick_occurred".to_string();
1323 let desired_log_3 = "Rebroadcasting monitor's pending claims".to_string();
1324 if log_entries.get(&("lightning_background_processor".to_string(), desired_log_1)).is_some() &&
1325 log_entries.get(&("lightning_background_processor".to_string(), desired_log_2)).is_some() &&
1326 log_entries.get(&("lightning_background_processor".to_string(), desired_log_3)).is_some() {
1331 if !std::thread::panicking() {
1332 bg_processor.stop().unwrap();
1337 fn test_channel_manager_persist_error() {
1338 // Test that if we encounter an error during manager persistence, the thread panics.
1339 let (_, nodes) = create_nodes(2, "test_persist_error");
1340 open_channel!(nodes[0], nodes[1], 100000);
1342 let data_dir = nodes[0].persister.get_data_dir();
1343 let persister = Arc::new(Persister::new(data_dir).with_manager_error(std::io::ErrorKind::Other, "test"));
1344 let event_handler = |_: _| {};
1345 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()));
1346 match bg_processor.join() {
1347 Ok(_) => panic!("Expected error persisting manager"),
1349 assert_eq!(e.kind(), std::io::ErrorKind::Other);
1350 assert_eq!(e.get_ref().unwrap().to_string(), "test");
1356 #[cfg(feature = "futures")]
1357 async fn test_channel_manager_persist_error_async() {
1358 // Test that if we encounter an error during manager persistence, the thread panics.
1359 let (_, nodes) = create_nodes(2, "test_persist_error_sync");
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"));
1365 let bp_future = super::process_events_async(
1366 persister, |_: _| {async {}}, nodes[0].chain_monitor.clone(), nodes[0].node.clone(),
1367 nodes[0].rapid_gossip_sync(), nodes[0].peer_manager.clone(), nodes[0].logger.clone(),
1368 Some(nodes[0].scorer.clone()), move |dur: Duration| {
1369 Box::pin(async move {
1370 tokio::time::sleep(dur).await;
1375 match bp_future.await {
1376 Ok(_) => panic!("Expected error persisting manager"),
1378 assert_eq!(e.kind(), std::io::ErrorKind::Other);
1379 assert_eq!(e.get_ref().unwrap().to_string(), "test");
1385 fn test_network_graph_persist_error() {
1386 // Test that if we encounter an error during network graph persistence, an error gets returned.
1387 let (_, nodes) = create_nodes(2, "test_persist_network_graph_error");
1388 let data_dir = nodes[0].persister.get_data_dir();
1389 let persister = Arc::new(Persister::new(data_dir).with_graph_error(std::io::ErrorKind::Other, "test"));
1390 let event_handler = |_: _| {};
1391 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()));
1393 match bg_processor.stop() {
1394 Ok(_) => panic!("Expected error persisting network graph"),
1396 assert_eq!(e.kind(), std::io::ErrorKind::Other);
1397 assert_eq!(e.get_ref().unwrap().to_string(), "test");
1403 fn test_scorer_persist_error() {
1404 // Test that if we encounter an error during scorer persistence, an error gets returned.
1405 let (_, nodes) = create_nodes(2, "test_persist_scorer_error");
1406 let data_dir = nodes[0].persister.get_data_dir();
1407 let persister = Arc::new(Persister::new(data_dir).with_scorer_error(std::io::ErrorKind::Other, "test"));
1408 let event_handler = |_: _| {};
1409 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()));
1411 match bg_processor.stop() {
1412 Ok(_) => panic!("Expected error persisting scorer"),
1414 assert_eq!(e.kind(), std::io::ErrorKind::Other);
1415 assert_eq!(e.get_ref().unwrap().to_string(), "test");
1421 fn test_background_event_handling() {
1422 let (_, mut nodes) = create_nodes(2, "test_background_event_handling");
1423 let channel_value = 100000;
1424 let data_dir = nodes[0].persister.get_data_dir();
1425 let persister = Arc::new(Persister::new(data_dir.clone()));
1427 // Set up a background event handler for FundingGenerationReady events.
1428 let (funding_generation_send, funding_generation_recv) = std::sync::mpsc::sync_channel(1);
1429 let (channel_pending_send, channel_pending_recv) = std::sync::mpsc::sync_channel(1);
1430 let event_handler = move |event: Event| match event {
1431 Event::FundingGenerationReady { .. } => funding_generation_send.send(handle_funding_generation_ready!(event, channel_value)).unwrap(),
1432 Event::ChannelPending { .. } => channel_pending_send.send(()).unwrap(),
1433 Event::ChannelReady { .. } => {},
1434 _ => panic!("Unexpected event: {:?}", event),
1437 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()));
1439 // Open a channel and check that the FundingGenerationReady event was handled.
1440 begin_open_channel!(nodes[0], nodes[1], channel_value);
1441 let (temporary_channel_id, funding_tx) = funding_generation_recv
1442 .recv_timeout(Duration::from_secs(EVENT_DEADLINE))
1443 .expect("FundingGenerationReady not handled within deadline");
1444 nodes[0].node.funding_transaction_generated(&temporary_channel_id, &nodes[1].node.get_our_node_id(), funding_tx.clone()).unwrap();
1445 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()));
1446 get_event!(nodes[1], Event::ChannelPending);
1447 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()));
1448 let _ = channel_pending_recv.recv_timeout(Duration::from_secs(EVENT_DEADLINE))
1449 .expect("ChannelPending not handled within deadline");
1451 // Confirm the funding transaction.
1452 confirm_transaction(&mut nodes[0], &funding_tx);
1453 let as_funding = get_event_msg!(nodes[0], MessageSendEvent::SendChannelReady, nodes[1].node.get_our_node_id());
1454 confirm_transaction(&mut nodes[1], &funding_tx);
1455 let bs_funding = get_event_msg!(nodes[1], MessageSendEvent::SendChannelReady, nodes[0].node.get_our_node_id());
1456 nodes[0].node.handle_channel_ready(&nodes[1].node.get_our_node_id(), &bs_funding);
1457 let _as_channel_update = get_event_msg!(nodes[0], MessageSendEvent::SendChannelUpdate, nodes[1].node.get_our_node_id());
1458 nodes[1].node.handle_channel_ready(&nodes[0].node.get_our_node_id(), &as_funding);
1459 let _bs_channel_update = get_event_msg!(nodes[1], MessageSendEvent::SendChannelUpdate, nodes[0].node.get_our_node_id());
1461 if !std::thread::panicking() {
1462 bg_processor.stop().unwrap();
1465 // Set up a background event handler for SpendableOutputs events.
1466 let (sender, receiver) = std::sync::mpsc::sync_channel(1);
1467 let event_handler = move |event: Event| match event {
1468 Event::SpendableOutputs { .. } => sender.send(event).unwrap(),
1469 Event::ChannelReady { .. } => {},
1470 Event::ChannelClosed { .. } => {},
1471 _ => panic!("Unexpected event: {:?}", event),
1473 let persister = Arc::new(Persister::new(data_dir));
1474 let bg_processor = BackgroundProcessor::start(persister, event_handler, nodes[0].chain_monitor.clone(), nodes[0].node.clone(), nodes[0].no_gossip_sync(), nodes[0].peer_manager.clone(), nodes[0].logger.clone(), Some(nodes[0].scorer.clone()));
1476 // Force close the channel and check that the SpendableOutputs event was handled.
1477 nodes[0].node.force_close_broadcasting_latest_txn(&nodes[0].node.list_channels()[0].channel_id, &nodes[1].node.get_our_node_id()).unwrap();
1478 let commitment_tx = nodes[0].tx_broadcaster.txn_broadcasted.lock().unwrap().pop().unwrap();
1479 confirm_transaction_depth(&mut nodes[0], &commitment_tx, BREAKDOWN_TIMEOUT as u32);
1481 let event = receiver
1482 .recv_timeout(Duration::from_secs(EVENT_DEADLINE))
1483 .expect("Events not handled within deadline");
1485 Event::SpendableOutputs { .. } => {},
1486 _ => panic!("Unexpected event: {:?}", event),
1489 if !std::thread::panicking() {
1490 bg_processor.stop().unwrap();
1495 fn test_scorer_persistence() {
1496 let (_, nodes) = create_nodes(2, "test_scorer_persistence");
1497 let data_dir = nodes[0].persister.get_data_dir();
1498 let persister = Arc::new(Persister::new(data_dir));
1499 let event_handler = |_: _| {};
1500 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()));
1503 let log_entries = nodes[0].logger.lines.lock().unwrap();
1504 let expected_log = "Persisting scorer".to_string();
1505 if log_entries.get(&("lightning_background_processor".to_string(), expected_log)).is_some() {
1510 if !std::thread::panicking() {
1511 bg_processor.stop().unwrap();
1515 macro_rules! do_test_not_pruning_network_graph_until_graph_sync_completion {
1516 ($nodes: expr, $receive: expr, $sleep: expr) => {
1517 let features = ChannelFeatures::empty();
1518 $nodes[0].network_graph.add_channel_from_partial_announcement(
1519 42, 53, features, $nodes[0].node.get_our_node_id(), $nodes[1].node.get_our_node_id()
1520 ).expect("Failed to update channel from partial announcement");
1521 let original_graph_description = $nodes[0].network_graph.to_string();
1522 assert!(original_graph_description.contains("42: features: 0000, node_one:"));
1523 assert_eq!($nodes[0].network_graph.read_only().channels().len(), 1);
1527 let log_entries = $nodes[0].logger.lines.lock().unwrap();
1528 let loop_counter = "Calling ChannelManager's timer_tick_occurred".to_string();
1529 if *log_entries.get(&("lightning_background_processor".to_string(), loop_counter))
1532 // Wait until the loop has gone around at least twice.
1537 let initialization_input = vec![
1538 76, 68, 75, 1, 111, 226, 140, 10, 182, 241, 179, 114, 193, 166, 162, 70, 174, 99, 247,
1539 79, 147, 30, 131, 101, 225, 90, 8, 156, 104, 214, 25, 0, 0, 0, 0, 0, 97, 227, 98, 218,
1540 0, 0, 0, 4, 2, 22, 7, 207, 206, 25, 164, 197, 231, 230, 231, 56, 102, 61, 250, 251,
1541 187, 172, 38, 46, 79, 247, 108, 44, 155, 48, 219, 238, 252, 53, 192, 6, 67, 2, 36, 125,
1542 157, 176, 223, 175, 234, 116, 94, 248, 201, 225, 97, 235, 50, 47, 115, 172, 63, 136,
1543 88, 216, 115, 11, 111, 217, 114, 84, 116, 124, 231, 107, 2, 158, 1, 242, 121, 152, 106,
1544 204, 131, 186, 35, 93, 70, 216, 10, 237, 224, 183, 89, 95, 65, 3, 83, 185, 58, 138,
1545 181, 64, 187, 103, 127, 68, 50, 2, 201, 19, 17, 138, 136, 149, 185, 226, 156, 137, 175,
1546 110, 32, 237, 0, 217, 90, 31, 100, 228, 149, 46, 219, 175, 168, 77, 4, 143, 38, 128,
1547 76, 97, 0, 0, 0, 2, 0, 0, 255, 8, 153, 192, 0, 2, 27, 0, 0, 0, 1, 0, 0, 255, 2, 68,
1548 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,
1549 0, 0, 0, 1, 0, 0, 0, 0, 58, 85, 116, 216, 255, 8, 153, 192, 0, 2, 27, 0, 0, 25, 0, 0,
1550 0, 1, 0, 0, 0, 125, 255, 2, 68, 226, 0, 6, 11, 0, 1, 5, 0, 0, 0, 0, 29, 129, 25, 192,
1552 $nodes[0].rapid_gossip_sync.update_network_graph_no_std(&initialization_input[..], Some(1642291930)).unwrap();
1554 // this should have added two channels and pruned the previous one.
1555 assert_eq!($nodes[0].network_graph.read_only().channels().len(), 2);
1557 $receive.expect("Network graph not pruned within deadline");
1559 // all channels should now be pruned
1560 assert_eq!($nodes[0].network_graph.read_only().channels().len(), 0);
1565 fn test_not_pruning_network_graph_until_graph_sync_completion() {
1566 let (sender, receiver) = std::sync::mpsc::sync_channel(1);
1568 let (_, nodes) = create_nodes(2, "test_not_pruning_network_graph_until_graph_sync_completion");
1569 let data_dir = nodes[0].persister.get_data_dir();
1570 let persister = Arc::new(Persister::new(data_dir).with_graph_persistence_notifier(sender));
1572 let event_handler = |_: _| {};
1573 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()));
1575 do_test_not_pruning_network_graph_until_graph_sync_completion!(nodes,
1576 receiver.recv_timeout(Duration::from_secs(super::FIRST_NETWORK_PRUNE_TIMER * 5)),
1577 std::thread::sleep(Duration::from_millis(1)));
1579 background_processor.stop().unwrap();
1583 #[cfg(feature = "futures")]
1584 async fn test_not_pruning_network_graph_until_graph_sync_completion_async() {
1585 let (sender, receiver) = std::sync::mpsc::sync_channel(1);
1587 let (_, nodes) = create_nodes(2, "test_not_pruning_network_graph_until_graph_sync_completion_async");
1588 let data_dir = nodes[0].persister.get_data_dir();
1589 let persister = Arc::new(Persister::new(data_dir).with_graph_persistence_notifier(sender));
1591 let (exit_sender, exit_receiver) = tokio::sync::watch::channel(());
1592 let bp_future = super::process_events_async(
1593 persister, |_: _| {async {}}, nodes[0].chain_monitor.clone(), nodes[0].node.clone(),
1594 nodes[0].rapid_gossip_sync(), nodes[0].peer_manager.clone(), nodes[0].logger.clone(),
1595 Some(nodes[0].scorer.clone()), move |dur: Duration| {
1596 let mut exit_receiver = exit_receiver.clone();
1597 Box::pin(async move {
1599 _ = tokio::time::sleep(dur) => false,
1600 _ = exit_receiver.changed() => true,
1606 let t1 = tokio::spawn(bp_future);
1607 let t2 = tokio::spawn(async move {
1608 do_test_not_pruning_network_graph_until_graph_sync_completion!(nodes, {
1611 tokio::time::sleep(Duration::from_secs(super::FIRST_NETWORK_PRUNE_TIMER)).await;
1612 if let Ok(()) = receiver.try_recv() { break Ok::<(), ()>(()); }
1616 }, tokio::time::sleep(Duration::from_millis(1)).await);
1617 exit_sender.send(()).unwrap();
1619 let (r1, r2) = tokio::join!(t1, t2);
1620 r1.unwrap().unwrap();
1624 macro_rules! do_test_payment_path_scoring {
1625 ($nodes: expr, $receive: expr) => {
1626 // Ensure that we update the scorer when relevant events are processed. In this case, we ensure
1627 // that we update the scorer upon a payment path succeeding (note that the channel must be
1628 // public or else we won't score it).
1629 // A background event handler for FundingGenerationReady events must be hooked up to a
1630 // running background processor.
1631 let scored_scid = 4242;
1632 let secp_ctx = Secp256k1::new();
1633 let node_1_privkey = SecretKey::from_slice(&[42; 32]).unwrap();
1634 let node_1_id = PublicKey::from_secret_key(&secp_ctx, &node_1_privkey);
1636 let path = Path { hops: vec![RouteHop {
1638 node_features: NodeFeatures::empty(),
1639 short_channel_id: scored_scid,
1640 channel_features: ChannelFeatures::empty(),
1642 cltv_expiry_delta: MIN_CLTV_EXPIRY_DELTA as u32,
1643 }], blinded_tail: None };
1645 $nodes[0].scorer.lock().unwrap().expect(TestResult::PaymentFailure { path: path.clone(), short_channel_id: scored_scid });
1646 $nodes[0].node.push_pending_event(Event::PaymentPathFailed {
1648 payment_hash: PaymentHash([42; 32]),
1649 payment_failed_permanently: false,
1650 failure: PathFailure::OnPath { network_update: None },
1652 short_channel_id: Some(scored_scid),
1654 let event = $receive.expect("PaymentPathFailed not handled within deadline");
1656 Event::PaymentPathFailed { .. } => {},
1657 _ => panic!("Unexpected event"),
1660 // Ensure we'll score payments that were explicitly failed back by the destination as
1662 $nodes[0].scorer.lock().unwrap().expect(TestResult::ProbeSuccess { path: path.clone() });
1663 $nodes[0].node.push_pending_event(Event::PaymentPathFailed {
1665 payment_hash: PaymentHash([42; 32]),
1666 payment_failed_permanently: true,
1667 failure: PathFailure::OnPath { network_update: None },
1669 short_channel_id: None,
1671 let event = $receive.expect("PaymentPathFailed not handled within deadline");
1673 Event::PaymentPathFailed { .. } => {},
1674 _ => panic!("Unexpected event"),
1677 $nodes[0].scorer.lock().unwrap().expect(TestResult::PaymentSuccess { path: path.clone() });
1678 $nodes[0].node.push_pending_event(Event::PaymentPathSuccessful {
1679 payment_id: PaymentId([42; 32]),
1683 let event = $receive.expect("PaymentPathSuccessful not handled within deadline");
1685 Event::PaymentPathSuccessful { .. } => {},
1686 _ => panic!("Unexpected event"),
1689 $nodes[0].scorer.lock().unwrap().expect(TestResult::ProbeSuccess { path: path.clone() });
1690 $nodes[0].node.push_pending_event(Event::ProbeSuccessful {
1691 payment_id: PaymentId([42; 32]),
1692 payment_hash: PaymentHash([42; 32]),
1695 let event = $receive.expect("ProbeSuccessful not handled within deadline");
1697 Event::ProbeSuccessful { .. } => {},
1698 _ => panic!("Unexpected event"),
1701 $nodes[0].scorer.lock().unwrap().expect(TestResult::ProbeFailure { path: path.clone() });
1702 $nodes[0].node.push_pending_event(Event::ProbeFailed {
1703 payment_id: PaymentId([42; 32]),
1704 payment_hash: PaymentHash([42; 32]),
1706 short_channel_id: Some(scored_scid),
1708 let event = $receive.expect("ProbeFailure not handled within deadline");
1710 Event::ProbeFailed { .. } => {},
1711 _ => panic!("Unexpected event"),
1717 fn test_payment_path_scoring() {
1718 let (sender, receiver) = std::sync::mpsc::sync_channel(1);
1719 let event_handler = move |event: Event| match event {
1720 Event::PaymentPathFailed { .. } => sender.send(event).unwrap(),
1721 Event::PaymentPathSuccessful { .. } => sender.send(event).unwrap(),
1722 Event::ProbeSuccessful { .. } => sender.send(event).unwrap(),
1723 Event::ProbeFailed { .. } => sender.send(event).unwrap(),
1724 _ => panic!("Unexpected event: {:?}", event),
1727 let (_, nodes) = create_nodes(1, "test_payment_path_scoring");
1728 let data_dir = nodes[0].persister.get_data_dir();
1729 let persister = Arc::new(Persister::new(data_dir));
1730 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()));
1732 do_test_payment_path_scoring!(nodes, receiver.recv_timeout(Duration::from_secs(EVENT_DEADLINE)));
1734 if !std::thread::panicking() {
1735 bg_processor.stop().unwrap();
1738 let log_entries = nodes[0].logger.lines.lock().unwrap();
1739 let expected_log = "Persisting scorer after update".to_string();
1740 assert_eq!(*log_entries.get(&("lightning_background_processor".to_string(), expected_log)).unwrap(), 5);
1744 #[cfg(feature = "futures")]
1745 async fn test_payment_path_scoring_async() {
1746 let (sender, mut receiver) = tokio::sync::mpsc::channel(1);
1747 let event_handler = move |event: Event| {
1748 let sender_ref = sender.clone();
1751 Event::PaymentPathFailed { .. } => { sender_ref.send(event).await.unwrap() },
1752 Event::PaymentPathSuccessful { .. } => { sender_ref.send(event).await.unwrap() },
1753 Event::ProbeSuccessful { .. } => { sender_ref.send(event).await.unwrap() },
1754 Event::ProbeFailed { .. } => { sender_ref.send(event).await.unwrap() },
1755 _ => panic!("Unexpected event: {:?}", event),
1760 let (_, nodes) = create_nodes(1, "test_payment_path_scoring_async");
1761 let data_dir = nodes[0].persister.get_data_dir();
1762 let persister = Arc::new(Persister::new(data_dir));
1764 let (exit_sender, exit_receiver) = tokio::sync::watch::channel(());
1766 let bp_future = super::process_events_async(
1767 persister, event_handler, nodes[0].chain_monitor.clone(), nodes[0].node.clone(),
1768 nodes[0].no_gossip_sync(), nodes[0].peer_manager.clone(), nodes[0].logger.clone(),
1769 Some(nodes[0].scorer.clone()), move |dur: Duration| {
1770 let mut exit_receiver = exit_receiver.clone();
1771 Box::pin(async move {
1773 _ = tokio::time::sleep(dur) => false,
1774 _ = exit_receiver.changed() => true,
1779 let t1 = tokio::spawn(bp_future);
1780 let t2 = tokio::spawn(async move {
1781 do_test_payment_path_scoring!(nodes, receiver.recv().await);
1782 exit_sender.send(()).unwrap();
1784 let log_entries = nodes[0].logger.lines.lock().unwrap();
1785 let expected_log = "Persisting scorer after update".to_string();
1786 assert_eq!(*log_entries.get(&("lightning_background_processor".to_string(), expected_log)).unwrap(), 5);
1789 let (r1, r2) = tokio::join!(t1, t2);
1790 r1.unwrap().unwrap();