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`] and [`PeerManager::timer_tick_occurred`]
68 /// 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;
119 #[cfg(feature = "futures")]
120 /// core::cmp::min is not currently const, so we define a trivial (and equivalent) replacement
121 const fn min_u64(a: u64, b: u64) -> u64 { if a < b { a } else { b } }
122 #[cfg(feature = "futures")]
123 const FASTEST_TIMER: u64 = min_u64(min_u64(FRESHNESS_TIMER, PING_TIMER),
124 min_u64(SCORER_PERSIST_TIMER, FIRST_NETWORK_PRUNE_TIMER));
126 /// Either [`P2PGossipSync`] or [`RapidGossipSync`].
128 P: Deref<Target = P2PGossipSync<G, U, L>>,
129 R: Deref<Target = RapidGossipSync<G, L>>,
130 G: Deref<Target = NetworkGraph<L>>,
134 where U::Target: UtxoLookup, L::Target: Logger {
135 /// Gossip sync via the lightning peer-to-peer network as defined by BOLT 7.
137 /// Rapid gossip sync from a trusted server.
144 P: Deref<Target = P2PGossipSync<G, U, L>>,
145 R: Deref<Target = RapidGossipSync<G, L>>,
146 G: Deref<Target = NetworkGraph<L>>,
149 > GossipSync<P, R, G, U, L>
150 where U::Target: UtxoLookup, L::Target: Logger {
151 fn network_graph(&self) -> Option<&G> {
153 GossipSync::P2P(gossip_sync) => Some(gossip_sync.network_graph()),
154 GossipSync::Rapid(gossip_sync) => Some(gossip_sync.network_graph()),
155 GossipSync::None => None,
159 fn prunable_network_graph(&self) -> Option<&G> {
161 GossipSync::P2P(gossip_sync) => Some(gossip_sync.network_graph()),
162 GossipSync::Rapid(gossip_sync) => {
163 if gossip_sync.is_initial_sync_complete() {
164 Some(gossip_sync.network_graph())
169 GossipSync::None => None,
174 /// This is not exported to bindings users as the bindings concretize everything and have constructors for us
175 impl<P: Deref<Target = P2PGossipSync<G, U, L>>, G: Deref<Target = NetworkGraph<L>>, U: Deref, L: Deref>
176 GossipSync<P, &RapidGossipSync<G, L>, G, U, L>
178 U::Target: UtxoLookup,
181 /// Initializes a new [`GossipSync::P2P`] variant.
182 pub fn p2p(gossip_sync: P) -> Self {
183 GossipSync::P2P(gossip_sync)
187 /// This is not exported to bindings users as the bindings concretize everything and have constructors for us
188 impl<'a, R: Deref<Target = RapidGossipSync<G, L>>, G: Deref<Target = NetworkGraph<L>>, L: Deref>
190 &P2PGossipSync<G, &'a (dyn UtxoLookup + Send + Sync), L>,
193 &'a (dyn UtxoLookup + Send + Sync),
199 /// Initializes a new [`GossipSync::Rapid`] variant.
200 pub fn rapid(gossip_sync: R) -> Self {
201 GossipSync::Rapid(gossip_sync)
205 /// This is not exported to bindings users as the bindings concretize everything and have constructors for us
208 &P2PGossipSync<&'a NetworkGraph<L>, &'a (dyn UtxoLookup + Send + Sync), L>,
209 &RapidGossipSync<&'a NetworkGraph<L>, L>,
211 &'a (dyn UtxoLookup + Send + Sync),
217 /// Initializes a new [`GossipSync::None`] variant.
218 pub fn none() -> Self {
223 fn handle_network_graph_update<L: Deref>(
224 network_graph: &NetworkGraph<L>, event: &Event
225 ) where L::Target: Logger {
226 if let Event::PaymentPathFailed {
227 failure: PathFailure::OnPath { network_update: Some(ref upd) }, .. } = event
229 network_graph.handle_network_update(upd);
233 fn update_scorer<'a, S: 'static + Deref<Target = SC> + Send + Sync, SC: 'a + WriteableScore<'a>>(
234 scorer: &'a S, event: &Event
236 let mut score = scorer.lock();
238 Event::PaymentPathFailed { ref path, short_channel_id: Some(scid), .. } => {
239 let path = path.iter().collect::<Vec<_>>();
240 score.payment_path_failed(&path, *scid);
242 Event::PaymentPathFailed { ref path, payment_failed_permanently: true, .. } => {
243 // Reached if the destination explicitly failed it back. We treat this as a successful probe
244 // because the payment made it all the way to the destination with sufficient liquidity.
245 let path = path.iter().collect::<Vec<_>>();
246 score.probe_successful(&path);
248 Event::PaymentPathSuccessful { path, .. } => {
249 let path = path.iter().collect::<Vec<_>>();
250 score.payment_path_successful(&path);
252 Event::ProbeSuccessful { path, .. } => {
253 let path = path.iter().collect::<Vec<_>>();
254 score.probe_successful(&path);
256 Event::ProbeFailed { path, short_channel_id: Some(scid), .. } => {
257 let path = path.iter().collect::<Vec<_>>();
258 score.probe_failed(&path, *scid);
264 macro_rules! define_run_body {
265 ($persister: ident, $chain_monitor: ident, $process_chain_monitor_events: expr,
266 $channel_manager: ident, $process_channel_manager_events: expr,
267 $gossip_sync: ident, $peer_manager: ident, $logger: ident, $scorer: ident,
268 $loop_exit_check: expr, $await: expr, $get_timer: expr, $timer_elapsed: expr,
269 $check_slow_await: expr)
271 log_trace!($logger, "Calling ChannelManager's timer_tick_occurred on startup");
272 $channel_manager.timer_tick_occurred();
274 let mut last_freshness_call = $get_timer(FRESHNESS_TIMER);
275 let mut last_ping_call = $get_timer(PING_TIMER);
276 let mut last_prune_call = $get_timer(FIRST_NETWORK_PRUNE_TIMER);
277 let mut last_scorer_persist_call = $get_timer(SCORER_PERSIST_TIMER);
278 let mut have_pruned = false;
281 $process_channel_manager_events;
282 $process_chain_monitor_events;
284 // Note that the PeerManager::process_events may block on ChannelManager's locks,
285 // hence it comes last here. When the ChannelManager finishes whatever it's doing,
286 // we want to ensure we get into `persist_manager` as quickly as we can, especially
287 // without running the normal event processing above and handing events to users.
289 // Specifically, on an *extremely* slow machine, we may see ChannelManager start
290 // processing a message effectively at any point during this loop. In order to
291 // minimize the time between such processing completing and persisting the updated
292 // ChannelManager, we want to minimize methods blocking on a ChannelManager
293 // generally, and as a fallback place such blocking only immediately before
295 $peer_manager.process_events();
297 // We wait up to 100ms, but track how long it takes to detect being put to sleep,
298 // see `await_start`'s use below.
299 let mut await_start = None;
300 if $check_slow_await { await_start = Some($get_timer(1)); }
301 let updates_available = $await;
302 let await_slow = if $check_slow_await { $timer_elapsed(&mut await_start.unwrap(), 1) } else { false };
304 if updates_available {
305 log_trace!($logger, "Persisting ChannelManager...");
306 $persister.persist_manager(&*$channel_manager)?;
307 log_trace!($logger, "Done persisting ChannelManager.");
309 // Exit the loop if the background processor was requested to stop.
310 if $loop_exit_check {
311 log_trace!($logger, "Terminating background processor.");
314 if $timer_elapsed(&mut last_freshness_call, FRESHNESS_TIMER) {
315 log_trace!($logger, "Calling ChannelManager's timer_tick_occurred");
316 $channel_manager.timer_tick_occurred();
317 last_freshness_call = $get_timer(FRESHNESS_TIMER);
320 // On various platforms, we may be starved of CPU cycles for several reasons.
321 // E.g. on iOS, if we've been in the background, we will be entirely paused.
322 // Similarly, if we're on a desktop platform and the device has been asleep, we
323 // may not get any cycles.
324 // We detect this by checking if our max-100ms-sleep, above, ran longer than a
325 // full second, at which point we assume sockets may have been killed (they
326 // appear to be at least on some platforms, even if it has only been a second).
327 // Note that we have to take care to not get here just because user event
328 // processing was slow at the top of the loop. For example, the sample client
329 // may call Bitcoin Core RPCs during event handling, which very often takes
330 // more than a handful of seconds to complete, and shouldn't disconnect all our
332 log_trace!($logger, "100ms sleep took more than a second, disconnecting peers.");
333 $peer_manager.disconnect_all_peers();
334 last_ping_call = $get_timer(PING_TIMER);
335 } else if $timer_elapsed(&mut last_ping_call, PING_TIMER) {
336 log_trace!($logger, "Calling PeerManager's timer_tick_occurred");
337 $peer_manager.timer_tick_occurred();
338 last_ping_call = $get_timer(PING_TIMER);
341 // Note that we want to run a graph prune once not long after startup before
342 // falling back to our usual hourly prunes. This avoids short-lived clients never
343 // pruning their network graph. We run once 60 seconds after startup before
344 // continuing our normal cadence.
345 if $timer_elapsed(&mut last_prune_call, if have_pruned { NETWORK_PRUNE_TIMER } else { FIRST_NETWORK_PRUNE_TIMER }) {
346 // The network graph must not be pruned while rapid sync completion is pending
347 if let Some(network_graph) = $gossip_sync.prunable_network_graph() {
348 #[cfg(feature = "std")] {
349 log_trace!($logger, "Pruning and persisting network graph.");
350 network_graph.remove_stale_channels_and_tracking();
352 #[cfg(not(feature = "std"))] {
353 log_warn!($logger, "Not pruning network graph, consider enabling `std` or doing so manually with remove_stale_channels_and_tracking_with_time.");
354 log_trace!($logger, "Persisting network graph.");
357 if let Err(e) = $persister.persist_graph(network_graph) {
358 log_error!($logger, "Error: Failed to persist network graph, check your disk and permissions {}", e)
363 last_prune_call = $get_timer(NETWORK_PRUNE_TIMER);
366 if $timer_elapsed(&mut last_scorer_persist_call, SCORER_PERSIST_TIMER) {
367 if let Some(ref scorer) = $scorer {
368 log_trace!($logger, "Persisting scorer");
369 if let Err(e) = $persister.persist_scorer(&scorer) {
370 log_error!($logger, "Error: Failed to persist scorer, check your disk and permissions {}", e)
373 last_scorer_persist_call = $get_timer(SCORER_PERSIST_TIMER);
377 // After we exit, ensure we persist the ChannelManager one final time - this avoids
378 // some races where users quit while channel updates were in-flight, with
379 // ChannelMonitor update(s) persisted without a corresponding ChannelManager update.
380 $persister.persist_manager(&*$channel_manager)?;
382 // Persist Scorer on exit
383 if let Some(ref scorer) = $scorer {
384 $persister.persist_scorer(&scorer)?;
387 // Persist NetworkGraph on exit
388 if let Some(network_graph) = $gossip_sync.network_graph() {
389 $persister.persist_graph(network_graph)?;
396 #[cfg(feature = "futures")]
397 pub(crate) mod futures_util {
398 use core::future::Future;
399 use core::task::{Poll, Waker, RawWaker, RawWakerVTable};
401 use core::marker::Unpin;
402 pub(crate) struct Selector<
403 A: Future<Output=()> + Unpin, B: Future<Output=()> + Unpin, C: Future<Output=bool> + Unpin
409 pub(crate) enum SelectorOutput {
414 A: Future<Output=()> + Unpin, B: Future<Output=()> + Unpin, C: Future<Output=bool> + Unpin
415 > Future for Selector<A, B, C> {
416 type Output = SelectorOutput;
417 fn poll(mut self: Pin<&mut Self>, ctx: &mut core::task::Context<'_>) -> Poll<SelectorOutput> {
418 match Pin::new(&mut self.a).poll(ctx) {
419 Poll::Ready(()) => { return Poll::Ready(SelectorOutput::A); },
422 match Pin::new(&mut self.b).poll(ctx) {
423 Poll::Ready(()) => { return Poll::Ready(SelectorOutput::B); },
426 match Pin::new(&mut self.c).poll(ctx) {
427 Poll::Ready(res) => { return Poll::Ready(SelectorOutput::C(res)); },
434 // If we want to poll a future without an async context to figure out if it has completed or
435 // not without awaiting, we need a Waker, which needs a vtable...we fill it with dummy values
436 // but sadly there's a good bit of boilerplate here.
437 fn dummy_waker_clone(_: *const ()) -> RawWaker { RawWaker::new(core::ptr::null(), &DUMMY_WAKER_VTABLE) }
438 fn dummy_waker_action(_: *const ()) { }
440 const DUMMY_WAKER_VTABLE: RawWakerVTable = RawWakerVTable::new(
441 dummy_waker_clone, dummy_waker_action, dummy_waker_action, dummy_waker_action);
442 pub(crate) fn dummy_waker() -> Waker { unsafe { Waker::from_raw(RawWaker::new(core::ptr::null(), &DUMMY_WAKER_VTABLE)) } }
444 #[cfg(feature = "futures")]
445 use futures_util::{Selector, SelectorOutput, dummy_waker};
446 #[cfg(feature = "futures")]
449 /// Processes background events in a future.
451 /// `sleeper` should return a future which completes in the given amount of time and returns a
452 /// boolean indicating whether the background processing should exit. Once `sleeper` returns a
453 /// future which outputs true, the loop will exit and this function's future will complete.
455 /// See [`BackgroundProcessor::start`] for information on which actions this handles.
457 /// Requires the `futures` feature. Note that while this method is available without the `std`
458 /// feature, doing so will skip calling [`NetworkGraph::remove_stale_channels_and_tracking`],
459 /// you should call [`NetworkGraph::remove_stale_channels_and_tracking_with_time`] regularly
460 /// manually instead.
462 /// The `mobile_interruptable_platform` flag should be set if we're currently running on a
463 /// mobile device, where we may need to check for interruption of the application regularly. If you
464 /// are unsure, you should set the flag, as the performance impact of it is minimal unless there
465 /// are hundreds or thousands of simultaneous process calls running.
466 #[cfg(feature = "futures")]
467 pub async fn process_events_async<
469 UL: 'static + Deref + Send + Sync,
470 CF: 'static + Deref + Send + Sync,
471 CW: 'static + Deref + Send + Sync,
472 T: 'static + Deref + Send + Sync,
473 ES: 'static + Deref + Send + Sync,
474 NS: 'static + Deref + Send + Sync,
475 SP: 'static + Deref + Send + Sync,
476 F: 'static + Deref + Send + Sync,
477 R: 'static + Deref + Send + Sync,
478 G: 'static + Deref<Target = NetworkGraph<L>> + Send + Sync,
479 L: 'static + Deref + Send + Sync,
480 P: 'static + Deref + Send + Sync,
481 Descriptor: 'static + SocketDescriptor + Send + Sync,
482 CMH: 'static + Deref + Send + Sync,
483 RMH: 'static + Deref + Send + Sync,
484 OMH: 'static + Deref + Send + Sync,
485 EventHandlerFuture: core::future::Future<Output = ()>,
486 EventHandler: Fn(Event) -> EventHandlerFuture,
487 PS: 'static + Deref + Send,
488 M: 'static + Deref<Target = ChainMonitor<<SP::Target as SignerProvider>::Signer, CF, T, F, L, P>> + Send + Sync,
489 CM: 'static + Deref<Target = ChannelManager<CW, T, ES, NS, SP, F, R, L>> + Send + Sync,
490 PGS: 'static + Deref<Target = P2PGossipSync<G, UL, L>> + Send + Sync,
491 RGS: 'static + Deref<Target = RapidGossipSync<G, L>> + Send,
492 UMH: 'static + Deref + Send + Sync,
493 PM: 'static + Deref<Target = PeerManager<Descriptor, CMH, RMH, OMH, L, UMH, NS>> + Send + Sync,
494 S: 'static + Deref<Target = SC> + Send + Sync,
495 SC: for<'b> WriteableScore<'b>,
496 SleepFuture: core::future::Future<Output = bool> + core::marker::Unpin,
497 Sleeper: Fn(Duration) -> SleepFuture
499 persister: PS, event_handler: EventHandler, chain_monitor: M, channel_manager: CM,
500 gossip_sync: GossipSync<PGS, RGS, G, UL, L>, peer_manager: PM, logger: L, scorer: Option<S>,
501 sleeper: Sleeper, mobile_interruptable_platform: bool,
502 ) -> Result<(), lightning::io::Error>
504 UL::Target: 'static + UtxoLookup,
505 CF::Target: 'static + chain::Filter,
506 CW::Target: 'static + chain::Watch<<SP::Target as SignerProvider>::Signer>,
507 T::Target: 'static + BroadcasterInterface,
508 ES::Target: 'static + EntropySource,
509 NS::Target: 'static + NodeSigner,
510 SP::Target: 'static + SignerProvider,
511 F::Target: 'static + FeeEstimator,
512 R::Target: 'static + Router,
513 L::Target: 'static + Logger,
514 P::Target: 'static + Persist<<SP::Target as SignerProvider>::Signer>,
515 CMH::Target: 'static + ChannelMessageHandler,
516 OMH::Target: 'static + OnionMessageHandler,
517 RMH::Target: 'static + RoutingMessageHandler,
518 UMH::Target: 'static + CustomMessageHandler,
519 PS::Target: 'static + Persister<'a, CW, T, ES, NS, SP, F, R, L, SC>,
521 let mut should_break = true;
522 let async_event_handler = |event| {
523 let network_graph = gossip_sync.network_graph();
524 let event_handler = &event_handler;
525 let scorer = &scorer;
527 if let Some(network_graph) = network_graph {
528 handle_network_graph_update(network_graph, &event)
530 if let Some(ref scorer) = scorer {
531 update_scorer(scorer, &event);
533 event_handler(event).await;
536 define_run_body!(persister,
537 chain_monitor, chain_monitor.process_pending_events_async(async_event_handler).await,
538 channel_manager, channel_manager.process_pending_events_async(async_event_handler).await,
539 gossip_sync, peer_manager, logger, scorer, should_break, {
541 a: channel_manager.get_persistable_update_future(),
542 b: chain_monitor.get_update_future(),
543 c: sleeper(if mobile_interruptable_platform { Duration::from_millis(100) } else { Duration::from_secs(FASTEST_TIMER) }),
546 SelectorOutput::A => true,
547 SelectorOutput::B => false,
548 SelectorOutput::C(exit) => {
553 }, |t| sleeper(Duration::from_secs(t)),
554 |fut: &mut SleepFuture, _| {
555 let mut waker = dummy_waker();
556 let mut ctx = task::Context::from_waker(&mut waker);
557 core::pin::Pin::new(fut).poll(&mut ctx).is_ready()
558 }, mobile_interruptable_platform)
561 #[cfg(feature = "std")]
562 impl BackgroundProcessor {
563 /// Start a background thread that takes care of responsibilities enumerated in the [top-level
566 /// The thread runs indefinitely unless the object is dropped, [`stop`] is called, or
567 /// [`Persister::persist_manager`] returns an error. In case of an error, the error is retrieved by calling
568 /// either [`join`] or [`stop`].
570 /// # Data Persistence
572 /// [`Persister::persist_manager`] is responsible for writing out the [`ChannelManager`] to disk, and/or
573 /// uploading to one or more backup services. See [`ChannelManager::write`] for writing out a
574 /// [`ChannelManager`]. See the `lightning-persister` crate for LDK's
575 /// provided implementation.
577 /// [`Persister::persist_graph`] is responsible for writing out the [`NetworkGraph`] to disk, if
578 /// [`GossipSync`] is supplied. See [`NetworkGraph::write`] for writing out a [`NetworkGraph`].
579 /// See the `lightning-persister` crate for LDK's provided implementation.
581 /// Typically, users should either implement [`Persister::persist_manager`] to never return an
582 /// error or call [`join`] and handle any error that may arise. For the latter case,
583 /// `BackgroundProcessor` must be restarted by calling `start` again after handling the error.
587 /// `event_handler` is responsible for handling events that users should be notified of (e.g.,
588 /// payment failed). [`BackgroundProcessor`] may decorate the given [`EventHandler`] with common
589 /// functionality implemented by other handlers.
590 /// * [`P2PGossipSync`] if given will update the [`NetworkGraph`] based on payment failures.
592 /// # Rapid Gossip Sync
594 /// If rapid gossip sync is meant to run at startup, pass [`RapidGossipSync`] via `gossip_sync`
595 /// to indicate that the [`BackgroundProcessor`] should not prune the [`NetworkGraph`] instance
596 /// until the [`RapidGossipSync`] instance completes its first sync.
598 /// [top-level documentation]: BackgroundProcessor
599 /// [`join`]: Self::join
600 /// [`stop`]: Self::stop
601 /// [`ChannelManager`]: lightning::ln::channelmanager::ChannelManager
602 /// [`ChannelManager::write`]: lightning::ln::channelmanager::ChannelManager#impl-Writeable
603 /// [`Persister::persist_manager`]: lightning::util::persist::Persister::persist_manager
604 /// [`Persister::persist_graph`]: lightning::util::persist::Persister::persist_graph
605 /// [`NetworkGraph`]: lightning::routing::gossip::NetworkGraph
606 /// [`NetworkGraph::write`]: lightning::routing::gossip::NetworkGraph#impl-Writeable
609 UL: 'static + Deref + Send + Sync,
610 CF: 'static + Deref + Send + Sync,
611 CW: 'static + Deref + Send + Sync,
612 T: 'static + Deref + Send + Sync,
613 ES: 'static + Deref + Send + Sync,
614 NS: 'static + Deref + Send + Sync,
615 SP: 'static + Deref + Send + Sync,
616 F: 'static + Deref + Send + Sync,
617 R: 'static + Deref + Send + Sync,
618 G: 'static + Deref<Target = NetworkGraph<L>> + Send + Sync,
619 L: 'static + Deref + Send + Sync,
620 P: 'static + Deref + Send + Sync,
621 Descriptor: 'static + SocketDescriptor + Send + Sync,
622 CMH: 'static + Deref + Send + Sync,
623 OMH: 'static + Deref + Send + Sync,
624 RMH: 'static + Deref + Send + Sync,
625 EH: 'static + EventHandler + Send,
626 PS: 'static + Deref + Send,
627 M: 'static + Deref<Target = ChainMonitor<<SP::Target as SignerProvider>::Signer, CF, T, F, L, P>> + Send + Sync,
628 CM: 'static + Deref<Target = ChannelManager<CW, T, ES, NS, SP, F, R, L>> + Send + Sync,
629 PGS: 'static + Deref<Target = P2PGossipSync<G, UL, L>> + Send + Sync,
630 RGS: 'static + Deref<Target = RapidGossipSync<G, L>> + Send,
631 UMH: 'static + Deref + Send + Sync,
632 PM: 'static + Deref<Target = PeerManager<Descriptor, CMH, RMH, OMH, L, UMH, NS>> + Send + Sync,
633 S: 'static + Deref<Target = SC> + Send + Sync,
634 SC: for <'b> WriteableScore<'b>,
636 persister: PS, event_handler: EH, chain_monitor: M, channel_manager: CM,
637 gossip_sync: GossipSync<PGS, RGS, G, UL, L>, peer_manager: PM, logger: L, scorer: Option<S>,
640 UL::Target: 'static + UtxoLookup,
641 CF::Target: 'static + chain::Filter,
642 CW::Target: 'static + chain::Watch<<SP::Target as SignerProvider>::Signer>,
643 T::Target: 'static + BroadcasterInterface,
644 ES::Target: 'static + EntropySource,
645 NS::Target: 'static + NodeSigner,
646 SP::Target: 'static + SignerProvider,
647 F::Target: 'static + FeeEstimator,
648 R::Target: 'static + Router,
649 L::Target: 'static + Logger,
650 P::Target: 'static + Persist<<SP::Target as SignerProvider>::Signer>,
651 CMH::Target: 'static + ChannelMessageHandler,
652 OMH::Target: 'static + OnionMessageHandler,
653 RMH::Target: 'static + RoutingMessageHandler,
654 UMH::Target: 'static + CustomMessageHandler,
655 PS::Target: 'static + Persister<'a, CW, T, ES, NS, SP, F, R, L, SC>,
657 let stop_thread = Arc::new(AtomicBool::new(false));
658 let stop_thread_clone = stop_thread.clone();
659 let handle = thread::spawn(move || -> Result<(), std::io::Error> {
660 let event_handler = |event| {
661 let network_graph = gossip_sync.network_graph();
662 if let Some(network_graph) = network_graph {
663 handle_network_graph_update(network_graph, &event)
665 if let Some(ref scorer) = scorer {
666 update_scorer(scorer, &event);
668 event_handler.handle_event(event);
670 define_run_body!(persister, chain_monitor, chain_monitor.process_pending_events(&event_handler),
671 channel_manager, channel_manager.process_pending_events(&event_handler),
672 gossip_sync, peer_manager, logger, scorer, stop_thread.load(Ordering::Acquire),
673 Sleeper::from_two_futures(
674 channel_manager.get_persistable_update_future(),
675 chain_monitor.get_update_future()
676 ).wait_timeout(Duration::from_millis(100)),
677 |_| Instant::now(), |time: &Instant, dur| time.elapsed().as_secs() > dur, false)
679 Self { stop_thread: stop_thread_clone, thread_handle: Some(handle) }
682 /// Join `BackgroundProcessor`'s thread, returning any error that occurred while persisting
683 /// [`ChannelManager`].
687 /// This function panics if the background thread has panicked such as while persisting or
690 /// [`ChannelManager`]: lightning::ln::channelmanager::ChannelManager
691 pub fn join(mut self) -> Result<(), std::io::Error> {
692 assert!(self.thread_handle.is_some());
696 /// Stop `BackgroundProcessor`'s thread, returning any error that occurred while persisting
697 /// [`ChannelManager`].
701 /// This function panics if the background thread has panicked such as while persisting or
704 /// [`ChannelManager`]: lightning::ln::channelmanager::ChannelManager
705 pub fn stop(mut self) -> Result<(), std::io::Error> {
706 assert!(self.thread_handle.is_some());
707 self.stop_and_join_thread()
710 fn stop_and_join_thread(&mut self) -> Result<(), std::io::Error> {
711 self.stop_thread.store(true, Ordering::Release);
715 fn join_thread(&mut self) -> Result<(), std::io::Error> {
716 match self.thread_handle.take() {
717 Some(handle) => handle.join().unwrap(),
723 #[cfg(feature = "std")]
724 impl Drop for BackgroundProcessor {
726 self.stop_and_join_thread().unwrap();
730 #[cfg(all(feature = "std", test))]
732 use bitcoin::blockdata::block::BlockHeader;
733 use bitcoin::blockdata::constants::genesis_block;
734 use bitcoin::blockdata::locktime::PackedLockTime;
735 use bitcoin::blockdata::transaction::{Transaction, TxOut};
736 use bitcoin::network::constants::Network;
737 use bitcoin::secp256k1::{SecretKey, PublicKey, Secp256k1};
738 use lightning::chain::{BestBlock, Confirm, chainmonitor};
739 use lightning::chain::channelmonitor::ANTI_REORG_DELAY;
740 use lightning::chain::keysinterface::{InMemorySigner, KeysManager};
741 use lightning::chain::transaction::OutPoint;
742 use lightning::events::{Event, PathFailure, MessageSendEventsProvider, MessageSendEvent};
743 use lightning::{get_event_msg, get_event};
744 use lightning::ln::PaymentHash;
745 use lightning::ln::channelmanager;
746 use lightning::ln::channelmanager::{BREAKDOWN_TIMEOUT, ChainParameters, MIN_CLTV_EXPIRY_DELTA, PaymentId};
747 use lightning::ln::features::{ChannelFeatures, NodeFeatures};
748 use lightning::ln::msgs::{ChannelMessageHandler, Init};
749 use lightning::ln::peer_handler::{PeerManager, MessageHandler, SocketDescriptor, IgnoringMessageHandler};
750 use lightning::routing::gossip::{NetworkGraph, NodeId, P2PGossipSync};
751 use lightning::routing::router::{DefaultRouter, RouteHop};
752 use lightning::routing::scoring::{ChannelUsage, Score};
753 use lightning::util::config::UserConfig;
754 use lightning::util::ser::Writeable;
755 use lightning::util::test_utils;
756 use lightning::util::persist::KVStorePersister;
757 use lightning_persister::FilesystemPersister;
758 use std::collections::VecDeque;
760 use std::path::PathBuf;
761 use std::sync::{Arc, Mutex};
762 use std::sync::mpsc::SyncSender;
763 use std::time::Duration;
764 use bitcoin::hashes::Hash;
765 use bitcoin::TxMerkleNode;
766 use lightning_rapid_gossip_sync::RapidGossipSync;
767 use super::{BackgroundProcessor, GossipSync, FRESHNESS_TIMER};
769 const EVENT_DEADLINE: u64 = 5 * FRESHNESS_TIMER;
771 #[derive(Clone, Hash, PartialEq, Eq)]
772 struct TestDescriptor{}
773 impl SocketDescriptor for TestDescriptor {
774 fn send_data(&mut self, _data: &[u8], _resume_read: bool) -> usize {
778 fn disconnect_socket(&mut self) {}
781 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>>;
783 type ChainMonitor = chainmonitor::ChainMonitor<InMemorySigner, Arc<test_utils::TestChainSource>, Arc<test_utils::TestBroadcaster>, Arc<test_utils::TestFeeEstimator>, Arc<test_utils::TestLogger>, Arc<FilesystemPersister>>;
785 type PGS = Arc<P2PGossipSync<Arc<NetworkGraph<Arc<test_utils::TestLogger>>>, Arc<test_utils::TestChainSource>, Arc<test_utils::TestLogger>>>;
786 type RGS = Arc<RapidGossipSync<Arc<NetworkGraph<Arc<test_utils::TestLogger>>>, Arc<test_utils::TestLogger>>>;
789 node: Arc<ChannelManager>,
790 p2p_gossip_sync: PGS,
791 rapid_gossip_sync: RGS,
792 peer_manager: Arc<PeerManager<TestDescriptor, Arc<test_utils::TestChannelMessageHandler>, Arc<test_utils::TestRoutingMessageHandler>, IgnoringMessageHandler, Arc<test_utils::TestLogger>, IgnoringMessageHandler, Arc<KeysManager>>>,
793 chain_monitor: Arc<ChainMonitor>,
794 persister: Arc<FilesystemPersister>,
795 tx_broadcaster: Arc<test_utils::TestBroadcaster>,
796 network_graph: Arc<NetworkGraph<Arc<test_utils::TestLogger>>>,
797 logger: Arc<test_utils::TestLogger>,
798 best_block: BestBlock,
799 scorer: Arc<Mutex<TestScorer>>,
803 fn p2p_gossip_sync(&self) -> GossipSync<PGS, RGS, Arc<NetworkGraph<Arc<test_utils::TestLogger>>>, Arc<test_utils::TestChainSource>, Arc<test_utils::TestLogger>> {
804 GossipSync::P2P(self.p2p_gossip_sync.clone())
807 fn rapid_gossip_sync(&self) -> GossipSync<PGS, RGS, Arc<NetworkGraph<Arc<test_utils::TestLogger>>>, Arc<test_utils::TestChainSource>, Arc<test_utils::TestLogger>> {
808 GossipSync::Rapid(self.rapid_gossip_sync.clone())
811 fn no_gossip_sync(&self) -> GossipSync<PGS, RGS, Arc<NetworkGraph<Arc<test_utils::TestLogger>>>, Arc<test_utils::TestChainSource>, Arc<test_utils::TestLogger>> {
818 let data_dir = self.persister.get_data_dir();
819 match fs::remove_dir_all(data_dir.clone()) {
820 Err(e) => println!("Failed to remove test persister directory {}: {}", data_dir, e),
827 graph_error: Option<(std::io::ErrorKind, &'static str)>,
828 graph_persistence_notifier: Option<SyncSender<()>>,
829 manager_error: Option<(std::io::ErrorKind, &'static str)>,
830 scorer_error: Option<(std::io::ErrorKind, &'static str)>,
831 filesystem_persister: FilesystemPersister,
835 fn new(data_dir: String) -> Self {
836 let filesystem_persister = FilesystemPersister::new(data_dir);
837 Self { graph_error: None, graph_persistence_notifier: None, manager_error: None, scorer_error: None, filesystem_persister }
840 fn with_graph_error(self, error: std::io::ErrorKind, message: &'static str) -> Self {
841 Self { graph_error: Some((error, message)), ..self }
844 fn with_graph_persistence_notifier(self, sender: SyncSender<()>) -> Self {
845 Self { graph_persistence_notifier: Some(sender), ..self }
848 fn with_manager_error(self, error: std::io::ErrorKind, message: &'static str) -> Self {
849 Self { manager_error: Some((error, message)), ..self }
852 fn with_scorer_error(self, error: std::io::ErrorKind, message: &'static str) -> Self {
853 Self { scorer_error: Some((error, message)), ..self }
857 impl KVStorePersister for Persister {
858 fn persist<W: Writeable>(&self, key: &str, object: &W) -> std::io::Result<()> {
859 if key == "manager" {
860 if let Some((error, message)) = self.manager_error {
861 return Err(std::io::Error::new(error, message))
865 if key == "network_graph" {
866 if let Some(sender) = &self.graph_persistence_notifier {
867 sender.send(()).unwrap();
870 if let Some((error, message)) = self.graph_error {
871 return Err(std::io::Error::new(error, message))
876 if let Some((error, message)) = self.scorer_error {
877 return Err(std::io::Error::new(error, message))
881 self.filesystem_persister.persist(key, object)
886 event_expectations: Option<VecDeque<TestResult>>,
891 PaymentFailure { path: Vec<RouteHop>, short_channel_id: u64 },
892 PaymentSuccess { path: Vec<RouteHop> },
893 ProbeFailure { path: Vec<RouteHop> },
894 ProbeSuccess { path: Vec<RouteHop> },
899 Self { event_expectations: None }
902 fn expect(&mut self, expectation: TestResult) {
903 self.event_expectations.get_or_insert_with(VecDeque::new).push_back(expectation);
907 impl lightning::util::ser::Writeable for TestScorer {
908 fn write<W: lightning::util::ser::Writer>(&self, _: &mut W) -> Result<(), lightning::io::Error> { Ok(()) }
911 impl Score for TestScorer {
912 fn channel_penalty_msat(
913 &self, _short_channel_id: u64, _source: &NodeId, _target: &NodeId, _usage: ChannelUsage
914 ) -> u64 { unimplemented!(); }
916 fn payment_path_failed(&mut self, actual_path: &[&RouteHop], actual_short_channel_id: u64) {
917 if let Some(expectations) = &mut self.event_expectations {
918 match expectations.pop_front().unwrap() {
919 TestResult::PaymentFailure { path, short_channel_id } => {
920 assert_eq!(actual_path, &path.iter().collect::<Vec<_>>()[..]);
921 assert_eq!(actual_short_channel_id, short_channel_id);
923 TestResult::PaymentSuccess { path } => {
924 panic!("Unexpected successful payment path: {:?}", path)
926 TestResult::ProbeFailure { path } => {
927 panic!("Unexpected probe failure: {:?}", path)
929 TestResult::ProbeSuccess { path } => {
930 panic!("Unexpected probe success: {:?}", path)
936 fn payment_path_successful(&mut self, actual_path: &[&RouteHop]) {
937 if let Some(expectations) = &mut self.event_expectations {
938 match expectations.pop_front().unwrap() {
939 TestResult::PaymentFailure { path, .. } => {
940 panic!("Unexpected payment path failure: {:?}", path)
942 TestResult::PaymentSuccess { path } => {
943 assert_eq!(actual_path, &path.iter().collect::<Vec<_>>()[..]);
945 TestResult::ProbeFailure { path } => {
946 panic!("Unexpected probe failure: {:?}", path)
948 TestResult::ProbeSuccess { path } => {
949 panic!("Unexpected probe success: {:?}", path)
955 fn probe_failed(&mut self, actual_path: &[&RouteHop], _: u64) {
956 if let Some(expectations) = &mut self.event_expectations {
957 match expectations.pop_front().unwrap() {
958 TestResult::PaymentFailure { path, .. } => {
959 panic!("Unexpected payment path failure: {:?}", path)
961 TestResult::PaymentSuccess { path } => {
962 panic!("Unexpected payment path success: {:?}", path)
964 TestResult::ProbeFailure { path } => {
965 assert_eq!(actual_path, &path.iter().collect::<Vec<_>>()[..]);
967 TestResult::ProbeSuccess { path } => {
968 panic!("Unexpected probe success: {:?}", path)
973 fn probe_successful(&mut self, actual_path: &[&RouteHop]) {
974 if let Some(expectations) = &mut self.event_expectations {
975 match expectations.pop_front().unwrap() {
976 TestResult::PaymentFailure { path, .. } => {
977 panic!("Unexpected payment path failure: {:?}", path)
979 TestResult::PaymentSuccess { path } => {
980 panic!("Unexpected payment path success: {:?}", path)
982 TestResult::ProbeFailure { path } => {
983 panic!("Unexpected probe failure: {:?}", path)
985 TestResult::ProbeSuccess { path } => {
986 assert_eq!(actual_path, &path.iter().collect::<Vec<_>>()[..]);
993 impl Drop for TestScorer {
995 if std::thread::panicking() {
999 if let Some(event_expectations) = &self.event_expectations {
1000 if !event_expectations.is_empty() {
1001 panic!("Unsatisfied event expectations: {:?}", event_expectations);
1007 fn get_full_filepath(filepath: String, filename: String) -> String {
1008 let mut path = PathBuf::from(filepath);
1009 path.push(filename);
1010 path.to_str().unwrap().to_string()
1013 fn create_nodes(num_nodes: usize, persist_dir: String) -> Vec<Node> {
1014 let mut nodes = Vec::new();
1015 for i in 0..num_nodes {
1016 let tx_broadcaster = Arc::new(test_utils::TestBroadcaster{txn_broadcasted: Mutex::new(Vec::new()), blocks: Arc::new(Mutex::new(Vec::new()))});
1017 let fee_estimator = Arc::new(test_utils::TestFeeEstimator { sat_per_kw: Mutex::new(253) });
1018 let logger = Arc::new(test_utils::TestLogger::with_id(format!("node {}", i)));
1019 let network = Network::Testnet;
1020 let genesis_block = genesis_block(network);
1021 let network_graph = Arc::new(NetworkGraph::new(network, logger.clone()));
1022 let scorer = Arc::new(Mutex::new(TestScorer::new()));
1023 let seed = [i as u8; 32];
1024 let router = Arc::new(DefaultRouter::new(network_graph.clone(), logger.clone(), seed, scorer.clone()));
1025 let chain_source = Arc::new(test_utils::TestChainSource::new(Network::Testnet));
1026 let persister = Arc::new(FilesystemPersister::new(format!("{}_persister_{}", persist_dir, i)));
1027 let now = Duration::from_secs(genesis_block.header.time as u64);
1028 let keys_manager = Arc::new(KeysManager::new(&seed, now.as_secs(), now.subsec_nanos()));
1029 let chain_monitor = Arc::new(chainmonitor::ChainMonitor::new(Some(chain_source.clone()), tx_broadcaster.clone(), logger.clone(), fee_estimator.clone(), persister.clone()));
1030 let best_block = BestBlock::from_network(network);
1031 let params = ChainParameters { network, best_block };
1032 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));
1033 let p2p_gossip_sync = Arc::new(P2PGossipSync::new(network_graph.clone(), Some(chain_source.clone()), logger.clone()));
1034 let rapid_gossip_sync = Arc::new(RapidGossipSync::new(network_graph.clone(), logger.clone()));
1035 let msg_handler = MessageHandler { chan_handler: Arc::new(test_utils::TestChannelMessageHandler::new()), route_handler: Arc::new(test_utils::TestRoutingMessageHandler::new()), onion_message_handler: IgnoringMessageHandler{}};
1036 let peer_manager = Arc::new(PeerManager::new(msg_handler, 0, &seed, logger.clone(), IgnoringMessageHandler{}, keys_manager.clone()));
1037 let node = Node { node: manager, p2p_gossip_sync, rapid_gossip_sync, peer_manager, chain_monitor, persister, tx_broadcaster, network_graph, logger, best_block, scorer };
1041 for i in 0..num_nodes {
1042 for j in (i+1)..num_nodes {
1043 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();
1044 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();
1051 macro_rules! open_channel {
1052 ($node_a: expr, $node_b: expr, $channel_value: expr) => {{
1053 begin_open_channel!($node_a, $node_b, $channel_value);
1054 let events = $node_a.node.get_and_clear_pending_events();
1055 assert_eq!(events.len(), 1);
1056 let (temporary_channel_id, tx) = handle_funding_generation_ready!(events[0], $channel_value);
1057 end_open_channel!($node_a, $node_b, temporary_channel_id, tx);
1062 macro_rules! begin_open_channel {
1063 ($node_a: expr, $node_b: expr, $channel_value: expr) => {{
1064 $node_a.node.create_channel($node_b.node.get_our_node_id(), $channel_value, 100, 42, None).unwrap();
1065 $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()));
1066 $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()));
1070 macro_rules! handle_funding_generation_ready {
1071 ($event: expr, $channel_value: expr) => {{
1073 Event::FundingGenerationReady { temporary_channel_id, channel_value_satoshis, ref output_script, user_channel_id, .. } => {
1074 assert_eq!(channel_value_satoshis, $channel_value);
1075 assert_eq!(user_channel_id, 42);
1077 let tx = Transaction { version: 1 as i32, lock_time: PackedLockTime(0), input: Vec::new(), output: vec![TxOut {
1078 value: channel_value_satoshis, script_pubkey: output_script.clone(),
1080 (temporary_channel_id, tx)
1082 _ => panic!("Unexpected event"),
1087 macro_rules! end_open_channel {
1088 ($node_a: expr, $node_b: expr, $temporary_channel_id: expr, $tx: expr) => {{
1089 $node_a.node.funding_transaction_generated(&$temporary_channel_id, &$node_b.node.get_our_node_id(), $tx.clone()).unwrap();
1090 $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()));
1091 get_event!($node_b, Event::ChannelPending);
1093 $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()));
1094 get_event!($node_a, Event::ChannelPending);
1098 fn confirm_transaction_depth(node: &mut Node, tx: &Transaction, depth: u32) {
1099 for i in 1..=depth {
1100 let prev_blockhash = node.best_block.block_hash();
1101 let height = node.best_block.height() + 1;
1102 let header = BlockHeader { version: 0x20000000, prev_blockhash, merkle_root: TxMerkleNode::all_zeros(), time: height, bits: 42, nonce: 42 };
1103 let txdata = vec![(0, tx)];
1104 node.best_block = BestBlock::new(header.block_hash(), height);
1107 node.node.transactions_confirmed(&header, &txdata, height);
1108 node.chain_monitor.transactions_confirmed(&header, &txdata, height);
1110 x if x == depth => {
1111 node.node.best_block_updated(&header, height);
1112 node.chain_monitor.best_block_updated(&header, height);
1118 fn confirm_transaction(node: &mut Node, tx: &Transaction) {
1119 confirm_transaction_depth(node, tx, ANTI_REORG_DELAY);
1123 fn test_background_processor() {
1124 // Test that when a new channel is created, the ChannelManager needs to be re-persisted with
1125 // updates. Also test that when new updates are available, the manager signals that it needs
1126 // re-persistence and is successfully re-persisted.
1127 let nodes = create_nodes(2, "test_background_processor".to_string());
1129 // Go through the channel creation process so that each node has something to persist. Since
1130 // open_channel consumes events, it must complete before starting BackgroundProcessor to
1131 // avoid a race with processing events.
1132 let tx = open_channel!(nodes[0], nodes[1], 100000);
1134 // Initiate the background processors to watch each node.
1135 let data_dir = nodes[0].persister.get_data_dir();
1136 let persister = Arc::new(Persister::new(data_dir));
1137 let event_handler = |_: _| {};
1138 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()));
1140 macro_rules! check_persisted_data {
1141 ($node: expr, $filepath: expr) => {
1142 let mut expected_bytes = Vec::new();
1144 expected_bytes.clear();
1145 match $node.write(&mut expected_bytes) {
1147 match std::fs::read($filepath) {
1149 if bytes == expected_bytes {
1158 Err(e) => panic!("Unexpected error: {}", e)
1164 // Check that the initial channel manager data is persisted as expected.
1165 let filepath = get_full_filepath("test_background_processor_persister_0".to_string(), "manager".to_string());
1166 check_persisted_data!(nodes[0].node, filepath.clone());
1169 if !nodes[0].node.get_persistence_condvar_value() { break }
1172 // Force-close the channel.
1173 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();
1175 // Check that the force-close updates are persisted.
1176 check_persisted_data!(nodes[0].node, filepath.clone());
1178 if !nodes[0].node.get_persistence_condvar_value() { break }
1181 // Check network graph is persisted
1182 let filepath = get_full_filepath("test_background_processor_persister_0".to_string(), "network_graph".to_string());
1183 check_persisted_data!(nodes[0].network_graph, filepath.clone());
1185 // Check scorer is persisted
1186 let filepath = get_full_filepath("test_background_processor_persister_0".to_string(), "scorer".to_string());
1187 check_persisted_data!(nodes[0].scorer, filepath.clone());
1189 assert!(bg_processor.stop().is_ok());
1193 fn test_timer_tick_called() {
1194 // Test that ChannelManager's and PeerManager's `timer_tick_occurred` is called every
1195 // `FRESHNESS_TIMER`.
1196 let nodes = create_nodes(1, "test_timer_tick_called".to_string());
1197 let data_dir = nodes[0].persister.get_data_dir();
1198 let persister = Arc::new(Persister::new(data_dir));
1199 let event_handler = |_: _| {};
1200 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()));
1202 let log_entries = nodes[0].logger.lines.lock().unwrap();
1203 let desired_log = "Calling ChannelManager's timer_tick_occurred".to_string();
1204 let second_desired_log = "Calling PeerManager's timer_tick_occurred".to_string();
1205 if log_entries.get(&("lightning_background_processor".to_string(), desired_log)).is_some() &&
1206 log_entries.get(&("lightning_background_processor".to_string(), second_desired_log)).is_some() {
1211 assert!(bg_processor.stop().is_ok());
1215 fn test_channel_manager_persist_error() {
1216 // Test that if we encounter an error during manager persistence, the thread panics.
1217 let nodes = create_nodes(2, "test_persist_error".to_string());
1218 open_channel!(nodes[0], nodes[1], 100000);
1220 let data_dir = nodes[0].persister.get_data_dir();
1221 let persister = Arc::new(Persister::new(data_dir).with_manager_error(std::io::ErrorKind::Other, "test"));
1222 let event_handler = |_: _| {};
1223 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()));
1224 match bg_processor.join() {
1225 Ok(_) => panic!("Expected error persisting manager"),
1227 assert_eq!(e.kind(), std::io::ErrorKind::Other);
1228 assert_eq!(e.get_ref().unwrap().to_string(), "test");
1234 fn test_network_graph_persist_error() {
1235 // Test that if we encounter an error during network graph persistence, an error gets returned.
1236 let nodes = create_nodes(2, "test_persist_network_graph_error".to_string());
1237 let data_dir = nodes[0].persister.get_data_dir();
1238 let persister = Arc::new(Persister::new(data_dir).with_graph_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].p2p_gossip_sync(), nodes[0].peer_manager.clone(), nodes[0].logger.clone(), Some(nodes[0].scorer.clone()));
1242 match bg_processor.stop() {
1243 Ok(_) => panic!("Expected error persisting network graph"),
1245 assert_eq!(e.kind(), std::io::ErrorKind::Other);
1246 assert_eq!(e.get_ref().unwrap().to_string(), "test");
1252 fn test_scorer_persist_error() {
1253 // Test that if we encounter an error during scorer persistence, an error gets returned.
1254 let nodes = create_nodes(2, "test_persist_scorer_error".to_string());
1255 let data_dir = nodes[0].persister.get_data_dir();
1256 let persister = Arc::new(Persister::new(data_dir).with_scorer_error(std::io::ErrorKind::Other, "test"));
1257 let event_handler = |_: _| {};
1258 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()));
1260 match bg_processor.stop() {
1261 Ok(_) => panic!("Expected error persisting scorer"),
1263 assert_eq!(e.kind(), std::io::ErrorKind::Other);
1264 assert_eq!(e.get_ref().unwrap().to_string(), "test");
1270 fn test_background_event_handling() {
1271 let mut nodes = create_nodes(2, "test_background_event_handling".to_string());
1272 let channel_value = 100000;
1273 let data_dir = nodes[0].persister.get_data_dir();
1274 let persister = Arc::new(Persister::new(data_dir.clone()));
1276 // Set up a background event handler for FundingGenerationReady events.
1277 let (sender, receiver) = std::sync::mpsc::sync_channel(1);
1278 let event_handler = move |event: Event| match event {
1279 Event::FundingGenerationReady { .. } => sender.send(handle_funding_generation_ready!(event, channel_value)).unwrap(),
1280 Event::ChannelReady { .. } => {},
1281 _ => panic!("Unexpected event: {:?}", event),
1284 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()));
1286 // Open a channel and check that the FundingGenerationReady event was handled.
1287 begin_open_channel!(nodes[0], nodes[1], channel_value);
1288 let (temporary_channel_id, funding_tx) = receiver
1289 .recv_timeout(Duration::from_secs(EVENT_DEADLINE))
1290 .expect("FundingGenerationReady not handled within deadline");
1291 end_open_channel!(nodes[0], nodes[1], temporary_channel_id, funding_tx);
1293 // Confirm the funding transaction.
1294 confirm_transaction(&mut nodes[0], &funding_tx);
1295 let as_funding = get_event_msg!(nodes[0], MessageSendEvent::SendChannelReady, nodes[1].node.get_our_node_id());
1296 confirm_transaction(&mut nodes[1], &funding_tx);
1297 let bs_funding = get_event_msg!(nodes[1], MessageSendEvent::SendChannelReady, nodes[0].node.get_our_node_id());
1298 nodes[0].node.handle_channel_ready(&nodes[1].node.get_our_node_id(), &bs_funding);
1299 let _as_channel_update = get_event_msg!(nodes[0], MessageSendEvent::SendChannelUpdate, nodes[1].node.get_our_node_id());
1300 nodes[1].node.handle_channel_ready(&nodes[0].node.get_our_node_id(), &as_funding);
1301 let _bs_channel_update = get_event_msg!(nodes[1], MessageSendEvent::SendChannelUpdate, nodes[0].node.get_our_node_id());
1303 assert!(bg_processor.stop().is_ok());
1305 // Set up a background event handler for SpendableOutputs events.
1306 let (sender, receiver) = std::sync::mpsc::sync_channel(1);
1307 let event_handler = move |event: Event| match event {
1308 Event::SpendableOutputs { .. } => sender.send(event).unwrap(),
1309 Event::ChannelReady { .. } => {},
1310 Event::ChannelClosed { .. } => {},
1311 _ => panic!("Unexpected event: {:?}", event),
1313 let persister = Arc::new(Persister::new(data_dir));
1314 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()));
1316 // Force close the channel and check that the SpendableOutputs event was handled.
1317 nodes[0].node.force_close_broadcasting_latest_txn(&nodes[0].node.list_channels()[0].channel_id, &nodes[1].node.get_our_node_id()).unwrap();
1318 let commitment_tx = nodes[0].tx_broadcaster.txn_broadcasted.lock().unwrap().pop().unwrap();
1319 confirm_transaction_depth(&mut nodes[0], &commitment_tx, BREAKDOWN_TIMEOUT as u32);
1321 let event = receiver
1322 .recv_timeout(Duration::from_secs(EVENT_DEADLINE))
1323 .expect("Events not handled within deadline");
1325 Event::SpendableOutputs { .. } => {},
1326 _ => panic!("Unexpected event: {:?}", event),
1329 assert!(bg_processor.stop().is_ok());
1333 fn test_scorer_persistence() {
1334 let nodes = create_nodes(2, "test_scorer_persistence".to_string());
1335 let data_dir = nodes[0].persister.get_data_dir();
1336 let persister = Arc::new(Persister::new(data_dir));
1337 let event_handler = |_: _| {};
1338 let bg_processor = BackgroundProcessor::start(persister, event_handler, nodes[0].chain_monitor.clone(), nodes[0].node.clone(), nodes[0].no_gossip_sync(), nodes[0].peer_manager.clone(), nodes[0].logger.clone(), Some(nodes[0].scorer.clone()));
1341 let log_entries = nodes[0].logger.lines.lock().unwrap();
1342 let expected_log = "Persisting scorer".to_string();
1343 if log_entries.get(&("lightning_background_processor".to_string(), expected_log)).is_some() {
1348 assert!(bg_processor.stop().is_ok());
1352 fn test_not_pruning_network_graph_until_graph_sync_completion() {
1353 let nodes = create_nodes(2, "test_not_pruning_network_graph_until_graph_sync_completion".to_string());
1354 let data_dir = nodes[0].persister.get_data_dir();
1355 let (sender, receiver) = std::sync::mpsc::sync_channel(1);
1356 let persister = Arc::new(Persister::new(data_dir).with_graph_persistence_notifier(sender));
1357 let network_graph = nodes[0].network_graph.clone();
1358 let features = ChannelFeatures::empty();
1359 network_graph.add_channel_from_partial_announcement(42, 53, features, nodes[0].node.get_our_node_id(), nodes[1].node.get_our_node_id())
1360 .expect("Failed to update channel from partial announcement");
1361 let original_graph_description = network_graph.to_string();
1362 assert!(original_graph_description.contains("42: features: 0000, node_one:"));
1363 assert_eq!(network_graph.read_only().channels().len(), 1);
1365 let event_handler = |_: _| {};
1366 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()));
1369 let log_entries = nodes[0].logger.lines.lock().unwrap();
1370 let loop_counter = "Calling ChannelManager's timer_tick_occurred".to_string();
1371 if *log_entries.get(&("lightning_background_processor".to_string(), loop_counter))
1374 // Wait until the loop has gone around at least twice.
1379 let initialization_input = vec![
1380 76, 68, 75, 1, 111, 226, 140, 10, 182, 241, 179, 114, 193, 166, 162, 70, 174, 99, 247,
1381 79, 147, 30, 131, 101, 225, 90, 8, 156, 104, 214, 25, 0, 0, 0, 0, 0, 97, 227, 98, 218,
1382 0, 0, 0, 4, 2, 22, 7, 207, 206, 25, 164, 197, 231, 230, 231, 56, 102, 61, 250, 251,
1383 187, 172, 38, 46, 79, 247, 108, 44, 155, 48, 219, 238, 252, 53, 192, 6, 67, 2, 36, 125,
1384 157, 176, 223, 175, 234, 116, 94, 248, 201, 225, 97, 235, 50, 47, 115, 172, 63, 136,
1385 88, 216, 115, 11, 111, 217, 114, 84, 116, 124, 231, 107, 2, 158, 1, 242, 121, 152, 106,
1386 204, 131, 186, 35, 93, 70, 216, 10, 237, 224, 183, 89, 95, 65, 3, 83, 185, 58, 138,
1387 181, 64, 187, 103, 127, 68, 50, 2, 201, 19, 17, 138, 136, 149, 185, 226, 156, 137, 175,
1388 110, 32, 237, 0, 217, 90, 31, 100, 228, 149, 46, 219, 175, 168, 77, 4, 143, 38, 128,
1389 76, 97, 0, 0, 0, 2, 0, 0, 255, 8, 153, 192, 0, 2, 27, 0, 0, 0, 1, 0, 0, 255, 2, 68,
1390 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,
1391 0, 0, 0, 1, 0, 0, 0, 0, 58, 85, 116, 216, 255, 8, 153, 192, 0, 2, 27, 0, 0, 25, 0, 0,
1392 0, 1, 0, 0, 0, 125, 255, 2, 68, 226, 0, 6, 11, 0, 1, 5, 0, 0, 0, 0, 29, 129, 25, 192,
1394 nodes[0].rapid_gossip_sync.update_network_graph_no_std(&initialization_input[..], Some(1642291930)).unwrap();
1396 // this should have added two channels
1397 assert_eq!(network_graph.read_only().channels().len(), 3);
1400 .recv_timeout(Duration::from_secs(super::FIRST_NETWORK_PRUNE_TIMER * 5))
1401 .expect("Network graph not pruned within deadline");
1403 background_processor.stop().unwrap();
1405 // all channels should now be pruned
1406 assert_eq!(network_graph.read_only().channels().len(), 0);
1410 fn test_payment_path_scoring() {
1411 // Ensure that we update the scorer when relevant events are processed. In this case, we ensure
1412 // that we update the scorer upon a payment path succeeding (note that the channel must be
1413 // public or else we won't score it).
1414 // Set up a background event handler for FundingGenerationReady events.
1415 let (sender, receiver) = std::sync::mpsc::sync_channel(1);
1416 let event_handler = move |event: Event| match event {
1417 Event::PaymentPathFailed { .. } => sender.send(event).unwrap(),
1418 Event::PaymentPathSuccessful { .. } => sender.send(event).unwrap(),
1419 Event::ProbeSuccessful { .. } => sender.send(event).unwrap(),
1420 Event::ProbeFailed { .. } => sender.send(event).unwrap(),
1421 _ => panic!("Unexpected event: {:?}", event),
1424 let nodes = create_nodes(1, "test_payment_path_scoring".to_string());
1425 let data_dir = nodes[0].persister.get_data_dir();
1426 let persister = Arc::new(Persister::new(data_dir));
1427 let bg_processor = BackgroundProcessor::start(persister, event_handler, nodes[0].chain_monitor.clone(), nodes[0].node.clone(), nodes[0].no_gossip_sync(), nodes[0].peer_manager.clone(), nodes[0].logger.clone(), Some(nodes[0].scorer.clone()));
1429 let scored_scid = 4242;
1430 let secp_ctx = Secp256k1::new();
1431 let node_1_privkey = SecretKey::from_slice(&[42; 32]).unwrap();
1432 let node_1_id = PublicKey::from_secret_key(&secp_ctx, &node_1_privkey);
1434 let path = vec![RouteHop {
1436 node_features: NodeFeatures::empty(),
1437 short_channel_id: scored_scid,
1438 channel_features: ChannelFeatures::empty(),
1440 cltv_expiry_delta: MIN_CLTV_EXPIRY_DELTA as u32,
1443 nodes[0].scorer.lock().unwrap().expect(TestResult::PaymentFailure { path: path.clone(), short_channel_id: scored_scid });
1444 nodes[0].node.push_pending_event(Event::PaymentPathFailed {
1446 payment_hash: PaymentHash([42; 32]),
1447 payment_failed_permanently: false,
1448 failure: PathFailure::OnPath { network_update: None },
1450 short_channel_id: Some(scored_scid),
1452 let event = receiver
1453 .recv_timeout(Duration::from_secs(EVENT_DEADLINE))
1454 .expect("PaymentPathFailed not handled within deadline");
1456 Event::PaymentPathFailed { .. } => {},
1457 _ => panic!("Unexpected event"),
1460 // Ensure we'll score payments that were explicitly failed back by the destination as
1462 nodes[0].scorer.lock().unwrap().expect(TestResult::ProbeSuccess { path: path.clone() });
1463 nodes[0].node.push_pending_event(Event::PaymentPathFailed {
1465 payment_hash: PaymentHash([42; 32]),
1466 payment_failed_permanently: true,
1467 failure: PathFailure::OnPath { network_update: None },
1469 short_channel_id: None,
1471 let event = receiver
1472 .recv_timeout(Duration::from_secs(EVENT_DEADLINE))
1473 .expect("PaymentPathFailed not handled within deadline");
1475 Event::PaymentPathFailed { .. } => {},
1476 _ => panic!("Unexpected event"),
1479 nodes[0].scorer.lock().unwrap().expect(TestResult::PaymentSuccess { path: path.clone() });
1480 nodes[0].node.push_pending_event(Event::PaymentPathSuccessful {
1481 payment_id: PaymentId([42; 32]),
1485 let event = receiver
1486 .recv_timeout(Duration::from_secs(EVENT_DEADLINE))
1487 .expect("PaymentPathSuccessful not handled within deadline");
1489 Event::PaymentPathSuccessful { .. } => {},
1490 _ => panic!("Unexpected event"),
1493 nodes[0].scorer.lock().unwrap().expect(TestResult::ProbeSuccess { path: path.clone() });
1494 nodes[0].node.push_pending_event(Event::ProbeSuccessful {
1495 payment_id: PaymentId([42; 32]),
1496 payment_hash: PaymentHash([42; 32]),
1499 let event = receiver
1500 .recv_timeout(Duration::from_secs(EVENT_DEADLINE))
1501 .expect("ProbeSuccessful not handled within deadline");
1503 Event::ProbeSuccessful { .. } => {},
1504 _ => panic!("Unexpected event"),
1507 nodes[0].scorer.lock().unwrap().expect(TestResult::ProbeFailure { path: path.clone() });
1508 nodes[0].node.push_pending_event(Event::ProbeFailed {
1509 payment_id: PaymentId([42; 32]),
1510 payment_hash: PaymentHash([42; 32]),
1512 short_channel_id: Some(scored_scid),
1514 let event = receiver
1515 .recv_timeout(Duration::from_secs(EVENT_DEADLINE))
1516 .expect("ProbeFailure not handled within deadline");
1518 Event::ProbeFailed { .. } => {},
1519 _ => panic!("Unexpected event"),
1522 assert!(bg_processor.stop().is_ok());