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 let path = path.iter().collect::<Vec<_>>();
245 score.payment_path_failed(&path, *scid);
247 Event::PaymentPathFailed { ref path, payment_failed_permanently: true, .. } => {
248 // Reached if the destination explicitly failed it back. We treat this as a successful probe
249 // because the payment made it all the way to the destination with sufficient liquidity.
250 let path = path.iter().collect::<Vec<_>>();
251 score.probe_successful(&path);
253 Event::PaymentPathSuccessful { path, .. } => {
254 let path = path.iter().collect::<Vec<_>>();
255 score.payment_path_successful(&path);
257 Event::ProbeSuccessful { path, .. } => {
258 let path = path.iter().collect::<Vec<_>>();
259 score.probe_successful(&path);
261 Event::ProbeFailed { path, short_channel_id: Some(scid), .. } => {
262 let path = path.iter().collect::<Vec<_>>();
263 score.probe_failed(&path, *scid);
269 macro_rules! define_run_body {
270 ($persister: ident, $chain_monitor: ident, $process_chain_monitor_events: expr,
271 $channel_manager: ident, $process_channel_manager_events: expr,
272 $gossip_sync: ident, $peer_manager: ident, $logger: ident, $scorer: ident,
273 $loop_exit_check: expr, $await: expr, $get_timer: expr, $timer_elapsed: expr,
274 $check_slow_await: expr)
276 log_trace!($logger, "Calling ChannelManager's timer_tick_occurred on startup");
277 $channel_manager.timer_tick_occurred();
278 log_trace!($logger, "Rebroadcasting monitor's pending claims on startup");
279 $chain_monitor.rebroadcast_pending_claims();
281 let mut last_freshness_call = $get_timer(FRESHNESS_TIMER);
282 let mut last_ping_call = $get_timer(PING_TIMER);
283 let mut last_prune_call = $get_timer(FIRST_NETWORK_PRUNE_TIMER);
284 let mut last_scorer_persist_call = $get_timer(SCORER_PERSIST_TIMER);
285 let mut last_rebroadcast_call = $get_timer(REBROADCAST_TIMER);
286 let mut have_pruned = false;
289 $process_channel_manager_events;
290 $process_chain_monitor_events;
292 // Note that the PeerManager::process_events may block on ChannelManager's locks,
293 // hence it comes last here. When the ChannelManager finishes whatever it's doing,
294 // we want to ensure we get into `persist_manager` as quickly as we can, especially
295 // without running the normal event processing above and handing events to users.
297 // Specifically, on an *extremely* slow machine, we may see ChannelManager start
298 // processing a message effectively at any point during this loop. In order to
299 // minimize the time between such processing completing and persisting the updated
300 // ChannelManager, we want to minimize methods blocking on a ChannelManager
301 // generally, and as a fallback place such blocking only immediately before
303 $peer_manager.process_events();
305 // We wait up to 100ms, but track how long it takes to detect being put to sleep,
306 // see `await_start`'s use below.
307 let mut await_start = None;
308 if $check_slow_await { await_start = Some($get_timer(1)); }
309 let updates_available = $await;
310 let await_slow = if $check_slow_await { $timer_elapsed(&mut await_start.unwrap(), 1) } else { false };
312 if updates_available {
313 log_trace!($logger, "Persisting ChannelManager...");
314 $persister.persist_manager(&*$channel_manager)?;
315 log_trace!($logger, "Done persisting ChannelManager.");
317 // Exit the loop if the background processor was requested to stop.
318 if $loop_exit_check {
319 log_trace!($logger, "Terminating background processor.");
322 if $timer_elapsed(&mut last_freshness_call, FRESHNESS_TIMER) {
323 log_trace!($logger, "Calling ChannelManager's timer_tick_occurred");
324 $channel_manager.timer_tick_occurred();
325 last_freshness_call = $get_timer(FRESHNESS_TIMER);
328 // On various platforms, we may be starved of CPU cycles for several reasons.
329 // E.g. on iOS, if we've been in the background, we will be entirely paused.
330 // Similarly, if we're on a desktop platform and the device has been asleep, we
331 // may not get any cycles.
332 // We detect this by checking if our max-100ms-sleep, above, ran longer than a
333 // full second, at which point we assume sockets may have been killed (they
334 // appear to be at least on some platforms, even if it has only been a second).
335 // Note that we have to take care to not get here just because user event
336 // processing was slow at the top of the loop. For example, the sample client
337 // may call Bitcoin Core RPCs during event handling, which very often takes
338 // more than a handful of seconds to complete, and shouldn't disconnect all our
340 log_trace!($logger, "100ms sleep took more than a second, disconnecting peers.");
341 $peer_manager.disconnect_all_peers();
342 last_ping_call = $get_timer(PING_TIMER);
343 } else if $timer_elapsed(&mut last_ping_call, PING_TIMER) {
344 log_trace!($logger, "Calling PeerManager's timer_tick_occurred");
345 $peer_manager.timer_tick_occurred();
346 last_ping_call = $get_timer(PING_TIMER);
349 // Note that we want to run a graph prune once not long after startup before
350 // falling back to our usual hourly prunes. This avoids short-lived clients never
351 // pruning their network graph. We run once 60 seconds after startup before
352 // continuing our normal cadence.
353 if $timer_elapsed(&mut last_prune_call, if have_pruned { NETWORK_PRUNE_TIMER } else { FIRST_NETWORK_PRUNE_TIMER }) {
354 // The network graph must not be pruned while rapid sync completion is pending
355 if let Some(network_graph) = $gossip_sync.prunable_network_graph() {
356 #[cfg(feature = "std")] {
357 log_trace!($logger, "Pruning and persisting network graph.");
358 network_graph.remove_stale_channels_and_tracking();
360 #[cfg(not(feature = "std"))] {
361 log_warn!($logger, "Not pruning network graph, consider enabling `std` or doing so manually with remove_stale_channels_and_tracking_with_time.");
362 log_trace!($logger, "Persisting network graph.");
365 if let Err(e) = $persister.persist_graph(network_graph) {
366 log_error!($logger, "Error: Failed to persist network graph, check your disk and permissions {}", e)
371 last_prune_call = $get_timer(NETWORK_PRUNE_TIMER);
374 if $timer_elapsed(&mut last_scorer_persist_call, SCORER_PERSIST_TIMER) {
375 if let Some(ref scorer) = $scorer {
376 log_trace!($logger, "Persisting scorer");
377 if let Err(e) = $persister.persist_scorer(&scorer) {
378 log_error!($logger, "Error: Failed to persist scorer, check your disk and permissions {}", e)
381 last_scorer_persist_call = $get_timer(SCORER_PERSIST_TIMER);
384 if $timer_elapsed(&mut last_rebroadcast_call, REBROADCAST_TIMER) {
385 log_trace!($logger, "Rebroadcasting monitor's pending claims");
386 $chain_monitor.rebroadcast_pending_claims();
387 last_rebroadcast_call = $get_timer(REBROADCAST_TIMER);
391 // After we exit, ensure we persist the ChannelManager one final time - this avoids
392 // some races where users quit while channel updates were in-flight, with
393 // ChannelMonitor update(s) persisted without a corresponding ChannelManager update.
394 $persister.persist_manager(&*$channel_manager)?;
396 // Persist Scorer on exit
397 if let Some(ref scorer) = $scorer {
398 $persister.persist_scorer(&scorer)?;
401 // Persist NetworkGraph on exit
402 if let Some(network_graph) = $gossip_sync.network_graph() {
403 $persister.persist_graph(network_graph)?;
410 #[cfg(feature = "futures")]
411 pub(crate) mod futures_util {
412 use core::future::Future;
413 use core::task::{Poll, Waker, RawWaker, RawWakerVTable};
415 use core::marker::Unpin;
416 pub(crate) struct Selector<
417 A: Future<Output=()> + Unpin, B: Future<Output=()> + Unpin, C: Future<Output=bool> + Unpin
423 pub(crate) enum SelectorOutput {
428 A: Future<Output=()> + Unpin, B: Future<Output=()> + Unpin, C: Future<Output=bool> + Unpin
429 > Future for Selector<A, B, C> {
430 type Output = SelectorOutput;
431 fn poll(mut self: Pin<&mut Self>, ctx: &mut core::task::Context<'_>) -> Poll<SelectorOutput> {
432 match Pin::new(&mut self.a).poll(ctx) {
433 Poll::Ready(()) => { return Poll::Ready(SelectorOutput::A); },
436 match Pin::new(&mut self.b).poll(ctx) {
437 Poll::Ready(()) => { return Poll::Ready(SelectorOutput::B); },
440 match Pin::new(&mut self.c).poll(ctx) {
441 Poll::Ready(res) => { return Poll::Ready(SelectorOutput::C(res)); },
448 // If we want to poll a future without an async context to figure out if it has completed or
449 // not without awaiting, we need a Waker, which needs a vtable...we fill it with dummy values
450 // but sadly there's a good bit of boilerplate here.
451 fn dummy_waker_clone(_: *const ()) -> RawWaker { RawWaker::new(core::ptr::null(), &DUMMY_WAKER_VTABLE) }
452 fn dummy_waker_action(_: *const ()) { }
454 const DUMMY_WAKER_VTABLE: RawWakerVTable = RawWakerVTable::new(
455 dummy_waker_clone, dummy_waker_action, dummy_waker_action, dummy_waker_action);
456 pub(crate) fn dummy_waker() -> Waker { unsafe { Waker::from_raw(RawWaker::new(core::ptr::null(), &DUMMY_WAKER_VTABLE)) } }
458 #[cfg(feature = "futures")]
459 use futures_util::{Selector, SelectorOutput, dummy_waker};
460 #[cfg(feature = "futures")]
463 /// Processes background events in a future.
465 /// `sleeper` should return a future which completes in the given amount of time and returns a
466 /// boolean indicating whether the background processing should exit. Once `sleeper` returns a
467 /// future which outputs true, the loop will exit and this function's future will complete.
469 /// See [`BackgroundProcessor::start`] for information on which actions this handles.
471 /// Requires the `futures` feature. Note that while this method is available without the `std`
472 /// feature, doing so will skip calling [`NetworkGraph::remove_stale_channels_and_tracking`],
473 /// you should call [`NetworkGraph::remove_stale_channels_and_tracking_with_time`] regularly
474 /// manually instead.
476 /// The `mobile_interruptable_platform` flag should be set if we're currently running on a
477 /// mobile device, where we may need to check for interruption of the application regularly. If you
478 /// are unsure, you should set the flag, as the performance impact of it is minimal unless there
479 /// are hundreds or thousands of simultaneous process calls running.
480 #[cfg(feature = "futures")]
481 pub async fn process_events_async<
483 UL: 'static + Deref + Send + Sync,
484 CF: 'static + Deref + Send + Sync,
485 CW: 'static + Deref + Send + Sync,
486 T: 'static + Deref + Send + Sync,
487 ES: 'static + Deref + Send + Sync,
488 NS: 'static + Deref + Send + Sync,
489 SP: 'static + Deref + Send + Sync,
490 F: 'static + Deref + Send + Sync,
491 R: 'static + Deref + Send + Sync,
492 G: 'static + Deref<Target = NetworkGraph<L>> + Send + Sync,
493 L: 'static + Deref + Send + Sync,
494 P: 'static + Deref + Send + Sync,
495 Descriptor: 'static + SocketDescriptor + Send + Sync,
496 CMH: 'static + Deref + Send + Sync,
497 RMH: 'static + Deref + Send + Sync,
498 OMH: 'static + Deref + Send + Sync,
499 EventHandlerFuture: core::future::Future<Output = ()>,
500 EventHandler: Fn(Event) -> EventHandlerFuture,
501 PS: 'static + Deref + Send,
502 M: 'static + Deref<Target = ChainMonitor<<SP::Target as SignerProvider>::Signer, CF, T, F, L, P>> + Send + Sync,
503 CM: 'static + Deref<Target = ChannelManager<CW, T, ES, NS, SP, F, R, L>> + Send + Sync,
504 PGS: 'static + Deref<Target = P2PGossipSync<G, UL, L>> + Send + Sync,
505 RGS: 'static + Deref<Target = RapidGossipSync<G, L>> + Send,
506 UMH: 'static + Deref + Send + Sync,
507 PM: 'static + Deref<Target = PeerManager<Descriptor, CMH, RMH, OMH, L, UMH, NS>> + Send + Sync,
508 S: 'static + Deref<Target = SC> + Send + Sync,
509 SC: for<'b> WriteableScore<'b>,
510 SleepFuture: core::future::Future<Output = bool> + core::marker::Unpin,
511 Sleeper: Fn(Duration) -> SleepFuture
513 persister: PS, event_handler: EventHandler, chain_monitor: M, channel_manager: CM,
514 gossip_sync: GossipSync<PGS, RGS, G, UL, L>, peer_manager: PM, logger: L, scorer: Option<S>,
515 sleeper: Sleeper, mobile_interruptable_platform: bool,
516 ) -> Result<(), lightning::io::Error>
518 UL::Target: 'static + UtxoLookup,
519 CF::Target: 'static + chain::Filter,
520 CW::Target: 'static + chain::Watch<<SP::Target as SignerProvider>::Signer>,
521 T::Target: 'static + BroadcasterInterface,
522 ES::Target: 'static + EntropySource,
523 NS::Target: 'static + NodeSigner,
524 SP::Target: 'static + SignerProvider,
525 F::Target: 'static + FeeEstimator,
526 R::Target: 'static + Router,
527 L::Target: 'static + Logger,
528 P::Target: 'static + Persist<<SP::Target as SignerProvider>::Signer>,
529 CMH::Target: 'static + ChannelMessageHandler,
530 OMH::Target: 'static + OnionMessageHandler,
531 RMH::Target: 'static + RoutingMessageHandler,
532 UMH::Target: 'static + CustomMessageHandler,
533 PS::Target: 'static + Persister<'a, CW, T, ES, NS, SP, F, R, L, SC>,
535 let mut should_break = false;
536 let async_event_handler = |event| {
537 let network_graph = gossip_sync.network_graph();
538 let event_handler = &event_handler;
539 let scorer = &scorer;
541 if let Some(network_graph) = network_graph {
542 handle_network_graph_update(network_graph, &event)
544 if let Some(ref scorer) = scorer {
545 update_scorer(scorer, &event);
547 event_handler(event).await;
550 define_run_body!(persister,
551 chain_monitor, chain_monitor.process_pending_events_async(async_event_handler).await,
552 channel_manager, channel_manager.process_pending_events_async(async_event_handler).await,
553 gossip_sync, peer_manager, logger, scorer, should_break, {
555 a: channel_manager.get_persistable_update_future(),
556 b: chain_monitor.get_update_future(),
557 c: sleeper(if mobile_interruptable_platform { Duration::from_millis(100) } else { Duration::from_secs(FASTEST_TIMER) }),
560 SelectorOutput::A => true,
561 SelectorOutput::B => false,
562 SelectorOutput::C(exit) => {
567 }, |t| sleeper(Duration::from_secs(t)),
568 |fut: &mut SleepFuture, _| {
569 let mut waker = dummy_waker();
570 let mut ctx = task::Context::from_waker(&mut waker);
571 match core::pin::Pin::new(fut).poll(&mut ctx) {
572 task::Poll::Ready(exit) => { should_break = exit; true },
573 task::Poll::Pending => false,
575 }, mobile_interruptable_platform)
578 #[cfg(feature = "std")]
579 impl BackgroundProcessor {
580 /// Start a background thread that takes care of responsibilities enumerated in the [top-level
583 /// The thread runs indefinitely unless the object is dropped, [`stop`] is called, or
584 /// [`Persister::persist_manager`] returns an error. In case of an error, the error is retrieved by calling
585 /// either [`join`] or [`stop`].
587 /// # Data Persistence
589 /// [`Persister::persist_manager`] is responsible for writing out the [`ChannelManager`] to disk, and/or
590 /// uploading to one or more backup services. See [`ChannelManager::write`] for writing out a
591 /// [`ChannelManager`]. See the `lightning-persister` crate for LDK's
592 /// provided implementation.
594 /// [`Persister::persist_graph`] is responsible for writing out the [`NetworkGraph`] to disk, if
595 /// [`GossipSync`] is supplied. See [`NetworkGraph::write`] for writing out a [`NetworkGraph`].
596 /// See the `lightning-persister` crate for LDK's provided implementation.
598 /// Typically, users should either implement [`Persister::persist_manager`] to never return an
599 /// error or call [`join`] and handle any error that may arise. For the latter case,
600 /// `BackgroundProcessor` must be restarted by calling `start` again after handling the error.
604 /// `event_handler` is responsible for handling events that users should be notified of (e.g.,
605 /// payment failed). [`BackgroundProcessor`] may decorate the given [`EventHandler`] with common
606 /// functionality implemented by other handlers.
607 /// * [`P2PGossipSync`] if given will update the [`NetworkGraph`] based on payment failures.
609 /// # Rapid Gossip Sync
611 /// If rapid gossip sync is meant to run at startup, pass [`RapidGossipSync`] via `gossip_sync`
612 /// to indicate that the [`BackgroundProcessor`] should not prune the [`NetworkGraph`] instance
613 /// until the [`RapidGossipSync`] instance completes its first sync.
615 /// [top-level documentation]: BackgroundProcessor
616 /// [`join`]: Self::join
617 /// [`stop`]: Self::stop
618 /// [`ChannelManager`]: lightning::ln::channelmanager::ChannelManager
619 /// [`ChannelManager::write`]: lightning::ln::channelmanager::ChannelManager#impl-Writeable
620 /// [`Persister::persist_manager`]: lightning::util::persist::Persister::persist_manager
621 /// [`Persister::persist_graph`]: lightning::util::persist::Persister::persist_graph
622 /// [`NetworkGraph`]: lightning::routing::gossip::NetworkGraph
623 /// [`NetworkGraph::write`]: lightning::routing::gossip::NetworkGraph#impl-Writeable
626 UL: 'static + Deref + Send + Sync,
627 CF: 'static + Deref + Send + Sync,
628 CW: 'static + Deref + Send + Sync,
629 T: 'static + Deref + Send + Sync,
630 ES: 'static + Deref + Send + Sync,
631 NS: 'static + Deref + Send + Sync,
632 SP: 'static + Deref + Send + Sync,
633 F: 'static + Deref + Send + Sync,
634 R: 'static + Deref + Send + Sync,
635 G: 'static + Deref<Target = NetworkGraph<L>> + Send + Sync,
636 L: 'static + Deref + Send + Sync,
637 P: 'static + Deref + Send + Sync,
638 Descriptor: 'static + SocketDescriptor + Send + Sync,
639 CMH: 'static + Deref + Send + Sync,
640 OMH: 'static + Deref + Send + Sync,
641 RMH: 'static + Deref + Send + Sync,
642 EH: 'static + EventHandler + Send,
643 PS: 'static + Deref + Send,
644 M: 'static + Deref<Target = ChainMonitor<<SP::Target as SignerProvider>::Signer, CF, T, F, L, P>> + Send + Sync,
645 CM: 'static + Deref<Target = ChannelManager<CW, T, ES, NS, SP, F, R, L>> + Send + Sync,
646 PGS: 'static + Deref<Target = P2PGossipSync<G, UL, L>> + Send + Sync,
647 RGS: 'static + Deref<Target = RapidGossipSync<G, L>> + Send,
648 UMH: 'static + Deref + Send + Sync,
649 PM: 'static + Deref<Target = PeerManager<Descriptor, CMH, RMH, OMH, L, UMH, NS>> + Send + Sync,
650 S: 'static + Deref<Target = SC> + Send + Sync,
651 SC: for <'b> WriteableScore<'b>,
653 persister: PS, event_handler: EH, chain_monitor: M, channel_manager: CM,
654 gossip_sync: GossipSync<PGS, RGS, G, UL, L>, peer_manager: PM, logger: L, scorer: Option<S>,
657 UL::Target: 'static + UtxoLookup,
658 CF::Target: 'static + chain::Filter,
659 CW::Target: 'static + chain::Watch<<SP::Target as SignerProvider>::Signer>,
660 T::Target: 'static + BroadcasterInterface,
661 ES::Target: 'static + EntropySource,
662 NS::Target: 'static + NodeSigner,
663 SP::Target: 'static + SignerProvider,
664 F::Target: 'static + FeeEstimator,
665 R::Target: 'static + Router,
666 L::Target: 'static + Logger,
667 P::Target: 'static + Persist<<SP::Target as SignerProvider>::Signer>,
668 CMH::Target: 'static + ChannelMessageHandler,
669 OMH::Target: 'static + OnionMessageHandler,
670 RMH::Target: 'static + RoutingMessageHandler,
671 UMH::Target: 'static + CustomMessageHandler,
672 PS::Target: 'static + Persister<'a, CW, T, ES, NS, SP, F, R, L, SC>,
674 let stop_thread = Arc::new(AtomicBool::new(false));
675 let stop_thread_clone = stop_thread.clone();
676 let handle = thread::spawn(move || -> Result<(), std::io::Error> {
677 let event_handler = |event| {
678 let network_graph = gossip_sync.network_graph();
679 if let Some(network_graph) = network_graph {
680 handle_network_graph_update(network_graph, &event)
682 if let Some(ref scorer) = scorer {
683 update_scorer(scorer, &event);
685 event_handler.handle_event(event);
687 define_run_body!(persister, chain_monitor, chain_monitor.process_pending_events(&event_handler),
688 channel_manager, channel_manager.process_pending_events(&event_handler),
689 gossip_sync, peer_manager, logger, scorer, stop_thread.load(Ordering::Acquire),
690 Sleeper::from_two_futures(
691 channel_manager.get_persistable_update_future(),
692 chain_monitor.get_update_future()
693 ).wait_timeout(Duration::from_millis(100)),
694 |_| Instant::now(), |time: &Instant, dur| time.elapsed().as_secs() > dur, false)
696 Self { stop_thread: stop_thread_clone, thread_handle: Some(handle) }
699 /// Join `BackgroundProcessor`'s thread, returning any error that occurred while persisting
700 /// [`ChannelManager`].
704 /// This function panics if the background thread has panicked such as while persisting or
707 /// [`ChannelManager`]: lightning::ln::channelmanager::ChannelManager
708 pub fn join(mut self) -> Result<(), std::io::Error> {
709 assert!(self.thread_handle.is_some());
713 /// Stop `BackgroundProcessor`'s thread, returning any error that occurred while persisting
714 /// [`ChannelManager`].
718 /// This function panics if the background thread has panicked such as while persisting or
721 /// [`ChannelManager`]: lightning::ln::channelmanager::ChannelManager
722 pub fn stop(mut self) -> Result<(), std::io::Error> {
723 assert!(self.thread_handle.is_some());
724 self.stop_and_join_thread()
727 fn stop_and_join_thread(&mut self) -> Result<(), std::io::Error> {
728 self.stop_thread.store(true, Ordering::Release);
732 fn join_thread(&mut self) -> Result<(), std::io::Error> {
733 match self.thread_handle.take() {
734 Some(handle) => handle.join().unwrap(),
740 #[cfg(feature = "std")]
741 impl Drop for BackgroundProcessor {
743 self.stop_and_join_thread().unwrap();
747 #[cfg(all(feature = "std", test))]
749 use bitcoin::blockdata::block::BlockHeader;
750 use bitcoin::blockdata::constants::genesis_block;
751 use bitcoin::blockdata::locktime::PackedLockTime;
752 use bitcoin::blockdata::transaction::{Transaction, TxOut};
753 use bitcoin::network::constants::Network;
754 use bitcoin::secp256k1::{SecretKey, PublicKey, Secp256k1};
755 use lightning::chain::{BestBlock, Confirm, chainmonitor};
756 use lightning::chain::channelmonitor::ANTI_REORG_DELAY;
757 use lightning::chain::keysinterface::{InMemorySigner, KeysManager};
758 use lightning::chain::transaction::OutPoint;
759 use lightning::events::{Event, PathFailure, MessageSendEventsProvider, MessageSendEvent};
760 use lightning::{get_event_msg, get_event};
761 use lightning::ln::PaymentHash;
762 use lightning::ln::channelmanager;
763 use lightning::ln::channelmanager::{BREAKDOWN_TIMEOUT, ChainParameters, MIN_CLTV_EXPIRY_DELTA, PaymentId};
764 use lightning::ln::features::{ChannelFeatures, NodeFeatures};
765 use lightning::ln::msgs::{ChannelMessageHandler, Init};
766 use lightning::ln::peer_handler::{PeerManager, MessageHandler, SocketDescriptor, IgnoringMessageHandler};
767 use lightning::routing::gossip::{NetworkGraph, NodeId, P2PGossipSync};
768 use lightning::routing::router::{DefaultRouter, RouteHop};
769 use lightning::routing::scoring::{ChannelUsage, Score};
770 use lightning::util::config::UserConfig;
771 use lightning::util::ser::Writeable;
772 use lightning::util::test_utils;
773 use lightning::util::persist::KVStorePersister;
774 use lightning_persister::FilesystemPersister;
775 use std::collections::VecDeque;
777 use std::path::PathBuf;
778 use std::sync::{Arc, Mutex};
779 use std::sync::mpsc::SyncSender;
780 use std::time::Duration;
781 use bitcoin::hashes::Hash;
782 use bitcoin::TxMerkleNode;
783 use lightning_rapid_gossip_sync::RapidGossipSync;
784 use super::{BackgroundProcessor, GossipSync, FRESHNESS_TIMER};
786 const EVENT_DEADLINE: u64 = 5 * FRESHNESS_TIMER;
788 #[derive(Clone, Hash, PartialEq, Eq)]
789 struct TestDescriptor{}
790 impl SocketDescriptor for TestDescriptor {
791 fn send_data(&mut self, _data: &[u8], _resume_read: bool) -> usize {
795 fn disconnect_socket(&mut self) {}
798 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>>;
800 type ChainMonitor = chainmonitor::ChainMonitor<InMemorySigner, Arc<test_utils::TestChainSource>, Arc<test_utils::TestBroadcaster>, Arc<test_utils::TestFeeEstimator>, Arc<test_utils::TestLogger>, Arc<FilesystemPersister>>;
802 type PGS = Arc<P2PGossipSync<Arc<NetworkGraph<Arc<test_utils::TestLogger>>>, Arc<test_utils::TestChainSource>, Arc<test_utils::TestLogger>>>;
803 type RGS = Arc<RapidGossipSync<Arc<NetworkGraph<Arc<test_utils::TestLogger>>>, Arc<test_utils::TestLogger>>>;
806 node: Arc<ChannelManager>,
807 p2p_gossip_sync: PGS,
808 rapid_gossip_sync: RGS,
809 peer_manager: Arc<PeerManager<TestDescriptor, Arc<test_utils::TestChannelMessageHandler>, Arc<test_utils::TestRoutingMessageHandler>, IgnoringMessageHandler, Arc<test_utils::TestLogger>, IgnoringMessageHandler, Arc<KeysManager>>>,
810 chain_monitor: Arc<ChainMonitor>,
811 persister: Arc<FilesystemPersister>,
812 tx_broadcaster: Arc<test_utils::TestBroadcaster>,
813 network_graph: Arc<NetworkGraph<Arc<test_utils::TestLogger>>>,
814 logger: Arc<test_utils::TestLogger>,
815 best_block: BestBlock,
816 scorer: Arc<Mutex<TestScorer>>,
820 fn p2p_gossip_sync(&self) -> GossipSync<PGS, RGS, Arc<NetworkGraph<Arc<test_utils::TestLogger>>>, Arc<test_utils::TestChainSource>, Arc<test_utils::TestLogger>> {
821 GossipSync::P2P(self.p2p_gossip_sync.clone())
824 fn rapid_gossip_sync(&self) -> GossipSync<PGS, RGS, Arc<NetworkGraph<Arc<test_utils::TestLogger>>>, Arc<test_utils::TestChainSource>, Arc<test_utils::TestLogger>> {
825 GossipSync::Rapid(self.rapid_gossip_sync.clone())
828 fn no_gossip_sync(&self) -> GossipSync<PGS, RGS, Arc<NetworkGraph<Arc<test_utils::TestLogger>>>, Arc<test_utils::TestChainSource>, Arc<test_utils::TestLogger>> {
835 let data_dir = self.persister.get_data_dir();
836 match fs::remove_dir_all(data_dir.clone()) {
837 Err(e) => println!("Failed to remove test persister directory {}: {}", data_dir, e),
844 graph_error: Option<(std::io::ErrorKind, &'static str)>,
845 graph_persistence_notifier: Option<SyncSender<()>>,
846 manager_error: Option<(std::io::ErrorKind, &'static str)>,
847 scorer_error: Option<(std::io::ErrorKind, &'static str)>,
848 filesystem_persister: FilesystemPersister,
852 fn new(data_dir: String) -> Self {
853 let filesystem_persister = FilesystemPersister::new(data_dir);
854 Self { graph_error: None, graph_persistence_notifier: None, manager_error: None, scorer_error: None, filesystem_persister }
857 fn with_graph_error(self, error: std::io::ErrorKind, message: &'static str) -> Self {
858 Self { graph_error: Some((error, message)), ..self }
861 fn with_graph_persistence_notifier(self, sender: SyncSender<()>) -> Self {
862 Self { graph_persistence_notifier: Some(sender), ..self }
865 fn with_manager_error(self, error: std::io::ErrorKind, message: &'static str) -> Self {
866 Self { manager_error: Some((error, message)), ..self }
869 fn with_scorer_error(self, error: std::io::ErrorKind, message: &'static str) -> Self {
870 Self { scorer_error: Some((error, message)), ..self }
874 impl KVStorePersister for Persister {
875 fn persist<W: Writeable>(&self, key: &str, object: &W) -> std::io::Result<()> {
876 if key == "manager" {
877 if let Some((error, message)) = self.manager_error {
878 return Err(std::io::Error::new(error, message))
882 if key == "network_graph" {
883 if let Some(sender) = &self.graph_persistence_notifier {
884 sender.send(()).unwrap();
887 if let Some((error, message)) = self.graph_error {
888 return Err(std::io::Error::new(error, message))
893 if let Some((error, message)) = self.scorer_error {
894 return Err(std::io::Error::new(error, message))
898 self.filesystem_persister.persist(key, object)
903 event_expectations: Option<VecDeque<TestResult>>,
908 PaymentFailure { path: Vec<RouteHop>, short_channel_id: u64 },
909 PaymentSuccess { path: Vec<RouteHop> },
910 ProbeFailure { path: Vec<RouteHop> },
911 ProbeSuccess { path: Vec<RouteHop> },
916 Self { event_expectations: None }
919 fn expect(&mut self, expectation: TestResult) {
920 self.event_expectations.get_or_insert_with(VecDeque::new).push_back(expectation);
924 impl lightning::util::ser::Writeable for TestScorer {
925 fn write<W: lightning::util::ser::Writer>(&self, _: &mut W) -> Result<(), lightning::io::Error> { Ok(()) }
928 impl Score for TestScorer {
929 fn channel_penalty_msat(
930 &self, _short_channel_id: u64, _source: &NodeId, _target: &NodeId, _usage: ChannelUsage
931 ) -> u64 { unimplemented!(); }
933 fn payment_path_failed(&mut self, actual_path: &[&RouteHop], actual_short_channel_id: u64) {
934 if let Some(expectations) = &mut self.event_expectations {
935 match expectations.pop_front().unwrap() {
936 TestResult::PaymentFailure { path, short_channel_id } => {
937 assert_eq!(actual_path, &path.iter().collect::<Vec<_>>()[..]);
938 assert_eq!(actual_short_channel_id, short_channel_id);
940 TestResult::PaymentSuccess { path } => {
941 panic!("Unexpected successful payment path: {:?}", path)
943 TestResult::ProbeFailure { path } => {
944 panic!("Unexpected probe failure: {:?}", path)
946 TestResult::ProbeSuccess { path } => {
947 panic!("Unexpected probe success: {:?}", path)
953 fn payment_path_successful(&mut self, actual_path: &[&RouteHop]) {
954 if let Some(expectations) = &mut self.event_expectations {
955 match expectations.pop_front().unwrap() {
956 TestResult::PaymentFailure { path, .. } => {
957 panic!("Unexpected payment path failure: {:?}", path)
959 TestResult::PaymentSuccess { path } => {
960 assert_eq!(actual_path, &path.iter().collect::<Vec<_>>()[..]);
962 TestResult::ProbeFailure { path } => {
963 panic!("Unexpected probe failure: {:?}", path)
965 TestResult::ProbeSuccess { path } => {
966 panic!("Unexpected probe success: {:?}", path)
972 fn probe_failed(&mut self, actual_path: &[&RouteHop], _: u64) {
973 if let Some(expectations) = &mut self.event_expectations {
974 match expectations.pop_front().unwrap() {
975 TestResult::PaymentFailure { path, .. } => {
976 panic!("Unexpected payment path failure: {:?}", path)
978 TestResult::PaymentSuccess { path } => {
979 panic!("Unexpected payment path success: {:?}", path)
981 TestResult::ProbeFailure { path } => {
982 assert_eq!(actual_path, &path.iter().collect::<Vec<_>>()[..]);
984 TestResult::ProbeSuccess { path } => {
985 panic!("Unexpected probe success: {:?}", path)
990 fn probe_successful(&mut self, actual_path: &[&RouteHop]) {
991 if let Some(expectations) = &mut self.event_expectations {
992 match expectations.pop_front().unwrap() {
993 TestResult::PaymentFailure { path, .. } => {
994 panic!("Unexpected payment path failure: {:?}", path)
996 TestResult::PaymentSuccess { path } => {
997 panic!("Unexpected payment path success: {:?}", path)
999 TestResult::ProbeFailure { path } => {
1000 panic!("Unexpected probe failure: {:?}", path)
1002 TestResult::ProbeSuccess { path } => {
1003 assert_eq!(actual_path, &path.iter().collect::<Vec<_>>()[..]);
1010 impl Drop for TestScorer {
1011 fn drop(&mut self) {
1012 if std::thread::panicking() {
1016 if let Some(event_expectations) = &self.event_expectations {
1017 if !event_expectations.is_empty() {
1018 panic!("Unsatisfied event expectations: {:?}", event_expectations);
1024 fn get_full_filepath(filepath: String, filename: String) -> String {
1025 let mut path = PathBuf::from(filepath);
1026 path.push(filename);
1027 path.to_str().unwrap().to_string()
1030 fn create_nodes(num_nodes: usize, persist_dir: String) -> Vec<Node> {
1031 let mut nodes = Vec::new();
1032 for i in 0..num_nodes {
1033 let tx_broadcaster = Arc::new(test_utils::TestBroadcaster{txn_broadcasted: Mutex::new(Vec::new()), blocks: Arc::new(Mutex::new(Vec::new()))});
1034 let fee_estimator = Arc::new(test_utils::TestFeeEstimator { sat_per_kw: Mutex::new(253) });
1035 let logger = Arc::new(test_utils::TestLogger::with_id(format!("node {}", i)));
1036 let network = Network::Testnet;
1037 let genesis_block = genesis_block(network);
1038 let network_graph = Arc::new(NetworkGraph::new(network, logger.clone()));
1039 let scorer = Arc::new(Mutex::new(TestScorer::new()));
1040 let seed = [i as u8; 32];
1041 let router = Arc::new(DefaultRouter::new(network_graph.clone(), logger.clone(), seed, scorer.clone()));
1042 let chain_source = Arc::new(test_utils::TestChainSource::new(Network::Testnet));
1043 let persister = Arc::new(FilesystemPersister::new(format!("{}_persister_{}", persist_dir, i)));
1044 let now = Duration::from_secs(genesis_block.header.time as u64);
1045 let keys_manager = Arc::new(KeysManager::new(&seed, now.as_secs(), now.subsec_nanos()));
1046 let chain_monitor = Arc::new(chainmonitor::ChainMonitor::new(Some(chain_source.clone()), tx_broadcaster.clone(), logger.clone(), fee_estimator.clone(), persister.clone()));
1047 let best_block = BestBlock::from_network(network);
1048 let params = ChainParameters { network, best_block };
1049 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));
1050 let p2p_gossip_sync = Arc::new(P2PGossipSync::new(network_graph.clone(), Some(chain_source.clone()), logger.clone()));
1051 let rapid_gossip_sync = Arc::new(RapidGossipSync::new(network_graph.clone(), logger.clone()));
1052 let msg_handler = MessageHandler { chan_handler: Arc::new(test_utils::TestChannelMessageHandler::new()), route_handler: Arc::new(test_utils::TestRoutingMessageHandler::new()), onion_message_handler: IgnoringMessageHandler{}};
1053 let peer_manager = Arc::new(PeerManager::new(msg_handler, 0, &seed, logger.clone(), IgnoringMessageHandler{}, keys_manager.clone()));
1054 let node = Node { node: manager, p2p_gossip_sync, rapid_gossip_sync, peer_manager, chain_monitor, persister, tx_broadcaster, network_graph, logger, best_block, scorer };
1058 for i in 0..num_nodes {
1059 for j in (i+1)..num_nodes {
1060 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();
1061 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();
1068 macro_rules! open_channel {
1069 ($node_a: expr, $node_b: expr, $channel_value: expr) => {{
1070 begin_open_channel!($node_a, $node_b, $channel_value);
1071 let events = $node_a.node.get_and_clear_pending_events();
1072 assert_eq!(events.len(), 1);
1073 let (temporary_channel_id, tx) = handle_funding_generation_ready!(events[0], $channel_value);
1074 $node_a.node.funding_transaction_generated(&temporary_channel_id, &$node_b.node.get_our_node_id(), tx.clone()).unwrap();
1075 $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()));
1076 get_event!($node_b, Event::ChannelPending);
1077 $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()));
1078 get_event!($node_a, Event::ChannelPending);
1083 macro_rules! begin_open_channel {
1084 ($node_a: expr, $node_b: expr, $channel_value: expr) => {{
1085 $node_a.node.create_channel($node_b.node.get_our_node_id(), $channel_value, 100, 42, None).unwrap();
1086 $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()));
1087 $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()));
1091 macro_rules! handle_funding_generation_ready {
1092 ($event: expr, $channel_value: expr) => {{
1094 Event::FundingGenerationReady { temporary_channel_id, channel_value_satoshis, ref output_script, user_channel_id, .. } => {
1095 assert_eq!(channel_value_satoshis, $channel_value);
1096 assert_eq!(user_channel_id, 42);
1098 let tx = Transaction { version: 1 as i32, lock_time: PackedLockTime(0), input: Vec::new(), output: vec![TxOut {
1099 value: channel_value_satoshis, script_pubkey: output_script.clone(),
1101 (temporary_channel_id, tx)
1103 _ => panic!("Unexpected event"),
1108 fn confirm_transaction_depth(node: &mut Node, tx: &Transaction, depth: u32) {
1109 for i in 1..=depth {
1110 let prev_blockhash = node.best_block.block_hash();
1111 let height = node.best_block.height() + 1;
1112 let header = BlockHeader { version: 0x20000000, prev_blockhash, merkle_root: TxMerkleNode::all_zeros(), time: height, bits: 42, nonce: 42 };
1113 let txdata = vec![(0, tx)];
1114 node.best_block = BestBlock::new(header.block_hash(), height);
1117 node.node.transactions_confirmed(&header, &txdata, height);
1118 node.chain_monitor.transactions_confirmed(&header, &txdata, height);
1120 x if x == depth => {
1121 node.node.best_block_updated(&header, height);
1122 node.chain_monitor.best_block_updated(&header, height);
1128 fn confirm_transaction(node: &mut Node, tx: &Transaction) {
1129 confirm_transaction_depth(node, tx, ANTI_REORG_DELAY);
1133 fn test_background_processor() {
1134 // Test that when a new channel is created, the ChannelManager needs to be re-persisted with
1135 // updates. Also test that when new updates are available, the manager signals that it needs
1136 // re-persistence and is successfully re-persisted.
1137 let nodes = create_nodes(2, "test_background_processor".to_string());
1139 // Go through the channel creation process so that each node has something to persist. Since
1140 // open_channel consumes events, it must complete before starting BackgroundProcessor to
1141 // avoid a race with processing events.
1142 let tx = open_channel!(nodes[0], nodes[1], 100000);
1144 // Initiate the background processors to watch each node.
1145 let data_dir = nodes[0].persister.get_data_dir();
1146 let persister = Arc::new(Persister::new(data_dir));
1147 let event_handler = |_: _| {};
1148 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()));
1150 macro_rules! check_persisted_data {
1151 ($node: expr, $filepath: expr) => {
1152 let mut expected_bytes = Vec::new();
1154 expected_bytes.clear();
1155 match $node.write(&mut expected_bytes) {
1157 match std::fs::read($filepath) {
1159 if bytes == expected_bytes {
1168 Err(e) => panic!("Unexpected error: {}", e)
1174 // Check that the initial channel manager data is persisted as expected.
1175 let filepath = get_full_filepath("test_background_processor_persister_0".to_string(), "manager".to_string());
1176 check_persisted_data!(nodes[0].node, filepath.clone());
1179 if !nodes[0].node.get_persistence_condvar_value() { break }
1182 // Force-close the channel.
1183 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();
1185 // Check that the force-close updates are persisted.
1186 check_persisted_data!(nodes[0].node, filepath.clone());
1188 if !nodes[0].node.get_persistence_condvar_value() { break }
1191 // Check network graph is persisted
1192 let filepath = get_full_filepath("test_background_processor_persister_0".to_string(), "network_graph".to_string());
1193 check_persisted_data!(nodes[0].network_graph, filepath.clone());
1195 // Check scorer is persisted
1196 let filepath = get_full_filepath("test_background_processor_persister_0".to_string(), "scorer".to_string());
1197 check_persisted_data!(nodes[0].scorer, filepath.clone());
1199 if !std::thread::panicking() {
1200 bg_processor.stop().unwrap();
1205 fn test_timer_tick_called() {
1206 // Test that `ChannelManager::timer_tick_occurred` is called every `FRESHNESS_TIMER`,
1207 // `ChainMonitor::rebroadcast_pending_claims` is called every `REBROADCAST_TIMER`, and
1208 // `PeerManager::timer_tick_occurred` every `PING_TIMER`.
1209 let nodes = create_nodes(1, "test_timer_tick_called".to_string());
1210 let data_dir = nodes[0].persister.get_data_dir();
1211 let persister = Arc::new(Persister::new(data_dir));
1212 let event_handler = |_: _| {};
1213 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()));
1215 let log_entries = nodes[0].logger.lines.lock().unwrap();
1216 let desired_log_1 = "Calling ChannelManager's timer_tick_occurred".to_string();
1217 let desired_log_2 = "Calling PeerManager's timer_tick_occurred".to_string();
1218 let desired_log_3 = "Rebroadcasting monitor's pending claims".to_string();
1219 if log_entries.get(&("lightning_background_processor".to_string(), desired_log_1)).is_some() &&
1220 log_entries.get(&("lightning_background_processor".to_string(), desired_log_2)).is_some() &&
1221 log_entries.get(&("lightning_background_processor".to_string(), desired_log_3)).is_some() {
1226 if !std::thread::panicking() {
1227 bg_processor.stop().unwrap();
1232 fn test_channel_manager_persist_error() {
1233 // Test that if we encounter an error during manager persistence, the thread panics.
1234 let nodes = create_nodes(2, "test_persist_error".to_string());
1235 open_channel!(nodes[0], nodes[1], 100000);
1237 let data_dir = nodes[0].persister.get_data_dir();
1238 let persister = Arc::new(Persister::new(data_dir).with_manager_error(std::io::ErrorKind::Other, "test"));
1239 let event_handler = |_: _| {};
1240 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()));
1241 match bg_processor.join() {
1242 Ok(_) => panic!("Expected error persisting manager"),
1244 assert_eq!(e.kind(), std::io::ErrorKind::Other);
1245 assert_eq!(e.get_ref().unwrap().to_string(), "test");
1251 #[cfg(feature = "futures")]
1252 async fn test_channel_manager_persist_error_async() {
1253 // Test that if we encounter an error during manager persistence, the thread panics.
1254 let nodes = create_nodes(2, "test_persist_error_sync".to_string());
1255 open_channel!(nodes[0], nodes[1], 100000);
1257 let data_dir = nodes[0].persister.get_data_dir();
1258 let persister = Arc::new(Persister::new(data_dir).with_manager_error(std::io::ErrorKind::Other, "test"));
1260 let bp_future = super::process_events_async(
1261 persister, |_: _| {async {}}, nodes[0].chain_monitor.clone(), nodes[0].node.clone(),
1262 nodes[0].rapid_gossip_sync(), nodes[0].peer_manager.clone(), nodes[0].logger.clone(),
1263 Some(nodes[0].scorer.clone()), move |dur: Duration| {
1264 Box::pin(async move {
1265 tokio::time::sleep(dur).await;
1270 match bp_future.await {
1271 Ok(_) => panic!("Expected error persisting manager"),
1273 assert_eq!(e.kind(), std::io::ErrorKind::Other);
1274 assert_eq!(e.get_ref().unwrap().to_string(), "test");
1280 fn test_network_graph_persist_error() {
1281 // Test that if we encounter an error during network graph persistence, an error gets returned.
1282 let nodes = create_nodes(2, "test_persist_network_graph_error".to_string());
1283 let data_dir = nodes[0].persister.get_data_dir();
1284 let persister = Arc::new(Persister::new(data_dir).with_graph_error(std::io::ErrorKind::Other, "test"));
1285 let event_handler = |_: _| {};
1286 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()));
1288 match bg_processor.stop() {
1289 Ok(_) => panic!("Expected error persisting network graph"),
1291 assert_eq!(e.kind(), std::io::ErrorKind::Other);
1292 assert_eq!(e.get_ref().unwrap().to_string(), "test");
1298 fn test_scorer_persist_error() {
1299 // Test that if we encounter an error during scorer persistence, an error gets returned.
1300 let nodes = create_nodes(2, "test_persist_scorer_error".to_string());
1301 let data_dir = nodes[0].persister.get_data_dir();
1302 let persister = Arc::new(Persister::new(data_dir).with_scorer_error(std::io::ErrorKind::Other, "test"));
1303 let event_handler = |_: _| {};
1304 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()));
1306 match bg_processor.stop() {
1307 Ok(_) => panic!("Expected error persisting scorer"),
1309 assert_eq!(e.kind(), std::io::ErrorKind::Other);
1310 assert_eq!(e.get_ref().unwrap().to_string(), "test");
1316 fn test_background_event_handling() {
1317 let mut nodes = create_nodes(2, "test_background_event_handling".to_string());
1318 let channel_value = 100000;
1319 let data_dir = nodes[0].persister.get_data_dir();
1320 let persister = Arc::new(Persister::new(data_dir.clone()));
1322 // Set up a background event handler for FundingGenerationReady events.
1323 let (funding_generation_send, funding_generation_recv) = std::sync::mpsc::sync_channel(1);
1324 let (channel_pending_send, channel_pending_recv) = std::sync::mpsc::sync_channel(1);
1325 let event_handler = move |event: Event| match event {
1326 Event::FundingGenerationReady { .. } => funding_generation_send.send(handle_funding_generation_ready!(event, channel_value)).unwrap(),
1327 Event::ChannelPending { .. } => channel_pending_send.send(()).unwrap(),
1328 Event::ChannelReady { .. } => {},
1329 _ => panic!("Unexpected event: {:?}", event),
1332 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()));
1334 // Open a channel and check that the FundingGenerationReady event was handled.
1335 begin_open_channel!(nodes[0], nodes[1], channel_value);
1336 let (temporary_channel_id, funding_tx) = funding_generation_recv
1337 .recv_timeout(Duration::from_secs(EVENT_DEADLINE))
1338 .expect("FundingGenerationReady not handled within deadline");
1339 nodes[0].node.funding_transaction_generated(&temporary_channel_id, &nodes[1].node.get_our_node_id(), funding_tx.clone()).unwrap();
1340 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()));
1341 get_event!(nodes[1], Event::ChannelPending);
1342 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()));
1343 let _ = channel_pending_recv.recv_timeout(Duration::from_secs(EVENT_DEADLINE))
1344 .expect("ChannelPending not handled within deadline");
1346 // Confirm the funding transaction.
1347 confirm_transaction(&mut nodes[0], &funding_tx);
1348 let as_funding = get_event_msg!(nodes[0], MessageSendEvent::SendChannelReady, nodes[1].node.get_our_node_id());
1349 confirm_transaction(&mut nodes[1], &funding_tx);
1350 let bs_funding = get_event_msg!(nodes[1], MessageSendEvent::SendChannelReady, nodes[0].node.get_our_node_id());
1351 nodes[0].node.handle_channel_ready(&nodes[1].node.get_our_node_id(), &bs_funding);
1352 let _as_channel_update = get_event_msg!(nodes[0], MessageSendEvent::SendChannelUpdate, nodes[1].node.get_our_node_id());
1353 nodes[1].node.handle_channel_ready(&nodes[0].node.get_our_node_id(), &as_funding);
1354 let _bs_channel_update = get_event_msg!(nodes[1], MessageSendEvent::SendChannelUpdate, nodes[0].node.get_our_node_id());
1356 if !std::thread::panicking() {
1357 bg_processor.stop().unwrap();
1360 // Set up a background event handler for SpendableOutputs events.
1361 let (sender, receiver) = std::sync::mpsc::sync_channel(1);
1362 let event_handler = move |event: Event| match event {
1363 Event::SpendableOutputs { .. } => sender.send(event).unwrap(),
1364 Event::ChannelReady { .. } => {},
1365 Event::ChannelClosed { .. } => {},
1366 _ => panic!("Unexpected event: {:?}", event),
1368 let persister = Arc::new(Persister::new(data_dir));
1369 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()));
1371 // Force close the channel and check that the SpendableOutputs event was handled.
1372 nodes[0].node.force_close_broadcasting_latest_txn(&nodes[0].node.list_channels()[0].channel_id, &nodes[1].node.get_our_node_id()).unwrap();
1373 let commitment_tx = nodes[0].tx_broadcaster.txn_broadcasted.lock().unwrap().pop().unwrap();
1374 confirm_transaction_depth(&mut nodes[0], &commitment_tx, BREAKDOWN_TIMEOUT as u32);
1376 let event = receiver
1377 .recv_timeout(Duration::from_secs(EVENT_DEADLINE))
1378 .expect("Events not handled within deadline");
1380 Event::SpendableOutputs { .. } => {},
1381 _ => panic!("Unexpected event: {:?}", event),
1384 if !std::thread::panicking() {
1385 bg_processor.stop().unwrap();
1390 fn test_scorer_persistence() {
1391 let nodes = create_nodes(2, "test_scorer_persistence".to_string());
1392 let data_dir = nodes[0].persister.get_data_dir();
1393 let persister = Arc::new(Persister::new(data_dir));
1394 let event_handler = |_: _| {};
1395 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()));
1398 let log_entries = nodes[0].logger.lines.lock().unwrap();
1399 let expected_log = "Persisting scorer".to_string();
1400 if log_entries.get(&("lightning_background_processor".to_string(), expected_log)).is_some() {
1405 if !std::thread::panicking() {
1406 bg_processor.stop().unwrap();
1410 macro_rules! do_test_not_pruning_network_graph_until_graph_sync_completion {
1411 ($nodes: expr, $receive: expr, $sleep: expr) => {
1412 let features = ChannelFeatures::empty();
1413 $nodes[0].network_graph.add_channel_from_partial_announcement(
1414 42, 53, features, $nodes[0].node.get_our_node_id(), $nodes[1].node.get_our_node_id()
1415 ).expect("Failed to update channel from partial announcement");
1416 let original_graph_description = $nodes[0].network_graph.to_string();
1417 assert!(original_graph_description.contains("42: features: 0000, node_one:"));
1418 assert_eq!($nodes[0].network_graph.read_only().channels().len(), 1);
1422 let log_entries = $nodes[0].logger.lines.lock().unwrap();
1423 let loop_counter = "Calling ChannelManager's timer_tick_occurred".to_string();
1424 if *log_entries.get(&("lightning_background_processor".to_string(), loop_counter))
1427 // Wait until the loop has gone around at least twice.
1432 let initialization_input = vec![
1433 76, 68, 75, 1, 111, 226, 140, 10, 182, 241, 179, 114, 193, 166, 162, 70, 174, 99, 247,
1434 79, 147, 30, 131, 101, 225, 90, 8, 156, 104, 214, 25, 0, 0, 0, 0, 0, 97, 227, 98, 218,
1435 0, 0, 0, 4, 2, 22, 7, 207, 206, 25, 164, 197, 231, 230, 231, 56, 102, 61, 250, 251,
1436 187, 172, 38, 46, 79, 247, 108, 44, 155, 48, 219, 238, 252, 53, 192, 6, 67, 2, 36, 125,
1437 157, 176, 223, 175, 234, 116, 94, 248, 201, 225, 97, 235, 50, 47, 115, 172, 63, 136,
1438 88, 216, 115, 11, 111, 217, 114, 84, 116, 124, 231, 107, 2, 158, 1, 242, 121, 152, 106,
1439 204, 131, 186, 35, 93, 70, 216, 10, 237, 224, 183, 89, 95, 65, 3, 83, 185, 58, 138,
1440 181, 64, 187, 103, 127, 68, 50, 2, 201, 19, 17, 138, 136, 149, 185, 226, 156, 137, 175,
1441 110, 32, 237, 0, 217, 90, 31, 100, 228, 149, 46, 219, 175, 168, 77, 4, 143, 38, 128,
1442 76, 97, 0, 0, 0, 2, 0, 0, 255, 8, 153, 192, 0, 2, 27, 0, 0, 0, 1, 0, 0, 255, 2, 68,
1443 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,
1444 0, 0, 0, 1, 0, 0, 0, 0, 58, 85, 116, 216, 255, 8, 153, 192, 0, 2, 27, 0, 0, 25, 0, 0,
1445 0, 1, 0, 0, 0, 125, 255, 2, 68, 226, 0, 6, 11, 0, 1, 5, 0, 0, 0, 0, 29, 129, 25, 192,
1447 $nodes[0].rapid_gossip_sync.update_network_graph_no_std(&initialization_input[..], Some(1642291930)).unwrap();
1449 // this should have added two channels and pruned the previous one.
1450 assert_eq!($nodes[0].network_graph.read_only().channels().len(), 2);
1452 $receive.expect("Network graph not pruned within deadline");
1454 // all channels should now be pruned
1455 assert_eq!($nodes[0].network_graph.read_only().channels().len(), 0);
1460 fn test_not_pruning_network_graph_until_graph_sync_completion() {
1461 let (sender, receiver) = std::sync::mpsc::sync_channel(1);
1463 let nodes = create_nodes(2, "test_not_pruning_network_graph_until_graph_sync_completion".to_string());
1464 let data_dir = nodes[0].persister.get_data_dir();
1465 let persister = Arc::new(Persister::new(data_dir).with_graph_persistence_notifier(sender));
1467 let event_handler = |_: _| {};
1468 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()));
1470 do_test_not_pruning_network_graph_until_graph_sync_completion!(nodes,
1471 receiver.recv_timeout(Duration::from_secs(super::FIRST_NETWORK_PRUNE_TIMER * 5)),
1472 std::thread::sleep(Duration::from_millis(1)));
1474 background_processor.stop().unwrap();
1478 #[cfg(feature = "futures")]
1479 async fn test_not_pruning_network_graph_until_graph_sync_completion_async() {
1480 let (sender, receiver) = std::sync::mpsc::sync_channel(1);
1482 let nodes = create_nodes(2, "test_not_pruning_network_graph_until_graph_sync_completion_async".to_string());
1483 let data_dir = nodes[0].persister.get_data_dir();
1484 let persister = Arc::new(Persister::new(data_dir).with_graph_persistence_notifier(sender));
1486 let (exit_sender, exit_receiver) = tokio::sync::watch::channel(());
1487 let bp_future = super::process_events_async(
1488 persister, |_: _| {async {}}, nodes[0].chain_monitor.clone(), nodes[0].node.clone(),
1489 nodes[0].rapid_gossip_sync(), nodes[0].peer_manager.clone(), nodes[0].logger.clone(),
1490 Some(nodes[0].scorer.clone()), move |dur: Duration| {
1491 let mut exit_receiver = exit_receiver.clone();
1492 Box::pin(async move {
1494 _ = tokio::time::sleep(dur) => false,
1495 _ = exit_receiver.changed() => true,
1500 // TODO: Drop _local and simply spawn after #2003
1501 let local_set = tokio::task::LocalSet::new();
1502 local_set.spawn_local(bp_future);
1503 local_set.spawn_local(async move {
1504 do_test_not_pruning_network_graph_until_graph_sync_completion!(nodes, {
1507 tokio::time::sleep(Duration::from_secs(super::FIRST_NETWORK_PRUNE_TIMER)).await;
1508 if let Ok(()) = receiver.try_recv() { break Ok::<(), ()>(()); }
1512 }, tokio::time::sleep(Duration::from_millis(1)).await);
1513 exit_sender.send(()).unwrap();
1518 macro_rules! do_test_payment_path_scoring {
1519 ($nodes: expr, $receive: expr) => {
1520 // Ensure that we update the scorer when relevant events are processed. In this case, we ensure
1521 // that we update the scorer upon a payment path succeeding (note that the channel must be
1522 // public or else we won't score it).
1523 // A background event handler for FundingGenerationReady events must be hooked up to a
1524 // running background processor.
1525 let scored_scid = 4242;
1526 let secp_ctx = Secp256k1::new();
1527 let node_1_privkey = SecretKey::from_slice(&[42; 32]).unwrap();
1528 let node_1_id = PublicKey::from_secret_key(&secp_ctx, &node_1_privkey);
1530 let path = vec![RouteHop {
1532 node_features: NodeFeatures::empty(),
1533 short_channel_id: scored_scid,
1534 channel_features: ChannelFeatures::empty(),
1536 cltv_expiry_delta: MIN_CLTV_EXPIRY_DELTA as u32,
1539 $nodes[0].scorer.lock().unwrap().expect(TestResult::PaymentFailure { path: path.clone(), short_channel_id: scored_scid });
1540 $nodes[0].node.push_pending_event(Event::PaymentPathFailed {
1542 payment_hash: PaymentHash([42; 32]),
1543 payment_failed_permanently: false,
1544 failure: PathFailure::OnPath { network_update: None },
1546 short_channel_id: Some(scored_scid),
1548 let event = $receive.expect("PaymentPathFailed not handled within deadline");
1550 Event::PaymentPathFailed { .. } => {},
1551 _ => panic!("Unexpected event"),
1554 // Ensure we'll score payments that were explicitly failed back by the destination as
1556 $nodes[0].scorer.lock().unwrap().expect(TestResult::ProbeSuccess { path: path.clone() });
1557 $nodes[0].node.push_pending_event(Event::PaymentPathFailed {
1559 payment_hash: PaymentHash([42; 32]),
1560 payment_failed_permanently: true,
1561 failure: PathFailure::OnPath { network_update: None },
1563 short_channel_id: None,
1565 let event = $receive.expect("PaymentPathFailed not handled within deadline");
1567 Event::PaymentPathFailed { .. } => {},
1568 _ => panic!("Unexpected event"),
1571 $nodes[0].scorer.lock().unwrap().expect(TestResult::PaymentSuccess { path: path.clone() });
1572 $nodes[0].node.push_pending_event(Event::PaymentPathSuccessful {
1573 payment_id: PaymentId([42; 32]),
1577 let event = $receive.expect("PaymentPathSuccessful not handled within deadline");
1579 Event::PaymentPathSuccessful { .. } => {},
1580 _ => panic!("Unexpected event"),
1583 $nodes[0].scorer.lock().unwrap().expect(TestResult::ProbeSuccess { path: path.clone() });
1584 $nodes[0].node.push_pending_event(Event::ProbeSuccessful {
1585 payment_id: PaymentId([42; 32]),
1586 payment_hash: PaymentHash([42; 32]),
1589 let event = $receive.expect("ProbeSuccessful not handled within deadline");
1591 Event::ProbeSuccessful { .. } => {},
1592 _ => panic!("Unexpected event"),
1595 $nodes[0].scorer.lock().unwrap().expect(TestResult::ProbeFailure { path: path.clone() });
1596 $nodes[0].node.push_pending_event(Event::ProbeFailed {
1597 payment_id: PaymentId([42; 32]),
1598 payment_hash: PaymentHash([42; 32]),
1600 short_channel_id: Some(scored_scid),
1602 let event = $receive.expect("ProbeFailure not handled within deadline");
1604 Event::ProbeFailed { .. } => {},
1605 _ => panic!("Unexpected event"),
1611 fn test_payment_path_scoring() {
1612 let (sender, receiver) = std::sync::mpsc::sync_channel(1);
1613 let event_handler = move |event: Event| match event {
1614 Event::PaymentPathFailed { .. } => sender.send(event).unwrap(),
1615 Event::PaymentPathSuccessful { .. } => sender.send(event).unwrap(),
1616 Event::ProbeSuccessful { .. } => sender.send(event).unwrap(),
1617 Event::ProbeFailed { .. } => sender.send(event).unwrap(),
1618 _ => panic!("Unexpected event: {:?}", event),
1621 let nodes = create_nodes(1, "test_payment_path_scoring".to_string());
1622 let data_dir = nodes[0].persister.get_data_dir();
1623 let persister = Arc::new(Persister::new(data_dir));
1624 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()));
1626 do_test_payment_path_scoring!(nodes, receiver.recv_timeout(Duration::from_secs(EVENT_DEADLINE)));
1628 if !std::thread::panicking() {
1629 bg_processor.stop().unwrap();
1634 #[cfg(feature = "futures")]
1635 async fn test_payment_path_scoring_async() {
1636 let (sender, mut receiver) = tokio::sync::mpsc::channel(1);
1637 let event_handler = move |event: Event| {
1638 let sender_ref = sender.clone();
1641 Event::PaymentPathFailed { .. } => { sender_ref.send(event).await.unwrap() },
1642 Event::PaymentPathSuccessful { .. } => { sender_ref.send(event).await.unwrap() },
1643 Event::ProbeSuccessful { .. } => { sender_ref.send(event).await.unwrap() },
1644 Event::ProbeFailed { .. } => { sender_ref.send(event).await.unwrap() },
1645 _ => panic!("Unexpected event: {:?}", event),
1650 let nodes = create_nodes(1, "test_payment_path_scoring_async".to_string());
1651 let data_dir = nodes[0].persister.get_data_dir();
1652 let persister = Arc::new(Persister::new(data_dir));
1654 let (exit_sender, exit_receiver) = tokio::sync::watch::channel(());
1656 let bp_future = super::process_events_async(
1657 persister, event_handler, nodes[0].chain_monitor.clone(), nodes[0].node.clone(),
1658 nodes[0].no_gossip_sync(), nodes[0].peer_manager.clone(), nodes[0].logger.clone(),
1659 Some(nodes[0].scorer.clone()), move |dur: Duration| {
1660 let mut exit_receiver = exit_receiver.clone();
1661 Box::pin(async move {
1663 _ = tokio::time::sleep(dur) => false,
1664 _ = exit_receiver.changed() => true,
1669 // TODO: Drop _local and simply spawn after #2003
1670 let local_set = tokio::task::LocalSet::new();
1671 local_set.spawn_local(bp_future);
1672 local_set.spawn_local(async move {
1673 do_test_payment_path_scoring!(nodes, receiver.recv().await);
1674 exit_sender.send(()).unwrap();