1 //! Utilities that take care of tasks that (1) need to happen periodically to keep Rust-Lightning
2 //! running properly, and (2) either can or should be run in the background. See docs for
3 //! [`BackgroundProcessor`] for more details on the nitty-gritty.
5 // Prefix these with `rustdoc::` when we update our MSRV to be >= 1.52 to remove warnings.
6 #![deny(broken_intra_doc_links)]
7 #![deny(private_intra_doc_links)]
10 #![cfg_attr(not(feature = "futures"), deny(unsafe_code))]
12 #![cfg_attr(docsrs, feature(doc_auto_cfg))]
14 #![cfg_attr(all(not(feature = "std"), not(test)), no_std)]
16 #[cfg(any(test, feature = "std"))]
19 #[cfg(not(feature = "std"))]
22 #[macro_use] extern crate lightning;
23 extern crate lightning_rapid_gossip_sync;
26 use lightning::chain::chaininterface::{BroadcasterInterface, FeeEstimator};
27 use lightning::chain::chainmonitor::{ChainMonitor, Persist};
28 use lightning::sign::{EntropySource, NodeSigner, SignerProvider};
29 use lightning::events::{Event, PathFailure};
30 #[cfg(feature = "std")]
31 use lightning::events::{EventHandler, EventsProvider};
32 use lightning::ln::channelmanager::ChannelManager;
33 use lightning::ln::peer_handler::APeerManager;
34 use lightning::routing::gossip::{NetworkGraph, P2PGossipSync};
35 use lightning::routing::utxo::UtxoLookup;
36 use lightning::routing::router::Router;
37 use lightning::routing::scoring::{ScoreUpdate, WriteableScore};
38 use lightning::util::logger::Logger;
39 use lightning::util::persist::Persister;
40 #[cfg(feature = "std")]
41 use lightning::util::wakers::Sleeper;
42 use lightning_rapid_gossip_sync::RapidGossipSync;
45 use core::time::Duration;
47 #[cfg(feature = "std")]
49 #[cfg(feature = "std")]
50 use core::sync::atomic::{AtomicBool, Ordering};
51 #[cfg(feature = "std")]
52 use std::thread::{self, JoinHandle};
53 #[cfg(feature = "std")]
54 use std::time::Instant;
56 #[cfg(not(feature = "std"))]
59 /// `BackgroundProcessor` takes care of tasks that (1) need to happen periodically to keep
60 /// Rust-Lightning running properly, and (2) either can or should be run in the background. Its
61 /// responsibilities are:
62 /// * Processing [`Event`]s with a user-provided [`EventHandler`].
63 /// * Monitoring whether the [`ChannelManager`] needs to be re-persisted to disk, and if so,
64 /// writing it to disk/backups by invoking the callback given to it at startup.
65 /// [`ChannelManager`] persistence should be done in the background.
66 /// * Calling [`ChannelManager::timer_tick_occurred`], [`ChainMonitor::rebroadcast_pending_claims`]
67 /// and [`PeerManager::timer_tick_occurred`] at the appropriate intervals.
68 /// * Calling [`NetworkGraph::remove_stale_channels_and_tracking`] (if a [`GossipSync`] with a
69 /// [`NetworkGraph`] is provided to [`BackgroundProcessor::start`]).
71 /// It will also call [`PeerManager::process_events`] periodically though this shouldn't be relied
72 /// upon as doing so may result in high latency.
76 /// If [`ChannelManager`] persistence fails and the persisted manager becomes out-of-date, then
77 /// there is a risk of channels force-closing on startup when the manager realizes it's outdated.
78 /// However, as long as [`ChannelMonitor`] backups are sound, no funds besides those used for
79 /// unilateral chain closure fees are at risk.
81 /// [`ChannelMonitor`]: lightning::chain::channelmonitor::ChannelMonitor
82 /// [`Event`]: lightning::events::Event
83 /// [`PeerManager::timer_tick_occurred`]: lightning::ln::peer_handler::PeerManager::timer_tick_occurred
84 /// [`PeerManager::process_events`]: lightning::ln::peer_handler::PeerManager::process_events
85 #[cfg(feature = "std")]
86 #[must_use = "BackgroundProcessor will immediately stop on drop. It should be stored until shutdown."]
87 pub struct BackgroundProcessor {
88 stop_thread: Arc<AtomicBool>,
89 thread_handle: Option<JoinHandle<Result<(), std::io::Error>>>,
93 const FRESHNESS_TIMER: u64 = 60;
95 const FRESHNESS_TIMER: u64 = 1;
97 #[cfg(all(not(test), not(debug_assertions)))]
98 const PING_TIMER: u64 = 10;
99 /// Signature operations take a lot longer without compiler optimisations.
100 /// Increasing the ping timer allows for this but slower devices will be disconnected if the
101 /// timeout is reached.
102 #[cfg(all(not(test), debug_assertions))]
103 const PING_TIMER: u64 = 30;
105 const PING_TIMER: u64 = 1;
107 /// Prune the network graph of stale entries hourly.
108 const NETWORK_PRUNE_TIMER: u64 = 60 * 60;
111 const SCORER_PERSIST_TIMER: u64 = 60 * 60;
113 const SCORER_PERSIST_TIMER: u64 = 1;
116 const FIRST_NETWORK_PRUNE_TIMER: u64 = 60;
118 const FIRST_NETWORK_PRUNE_TIMER: u64 = 1;
121 const REBROADCAST_TIMER: u64 = 30;
123 const REBROADCAST_TIMER: u64 = 1;
125 #[cfg(feature = "futures")]
126 /// core::cmp::min is not currently const, so we define a trivial (and equivalent) replacement
127 const fn min_u64(a: u64, b: u64) -> u64 { if a < b { a } else { b } }
128 #[cfg(feature = "futures")]
129 const FASTEST_TIMER: u64 = min_u64(min_u64(FRESHNESS_TIMER, PING_TIMER),
130 min_u64(SCORER_PERSIST_TIMER, min_u64(FIRST_NETWORK_PRUNE_TIMER, REBROADCAST_TIMER)));
132 /// Either [`P2PGossipSync`] or [`RapidGossipSync`].
134 P: Deref<Target = P2PGossipSync<G, U, L>>,
135 R: Deref<Target = RapidGossipSync<G, L>>,
136 G: Deref<Target = NetworkGraph<L>>,
140 where U::Target: UtxoLookup, L::Target: Logger {
141 /// Gossip sync via the lightning peer-to-peer network as defined by BOLT 7.
143 /// Rapid gossip sync from a trusted server.
150 P: Deref<Target = P2PGossipSync<G, U, L>>,
151 R: Deref<Target = RapidGossipSync<G, L>>,
152 G: Deref<Target = NetworkGraph<L>>,
155 > GossipSync<P, R, G, U, L>
156 where U::Target: UtxoLookup, L::Target: Logger {
157 fn network_graph(&self) -> Option<&G> {
159 GossipSync::P2P(gossip_sync) => Some(gossip_sync.network_graph()),
160 GossipSync::Rapid(gossip_sync) => Some(gossip_sync.network_graph()),
161 GossipSync::None => None,
165 fn prunable_network_graph(&self) -> Option<&G> {
167 GossipSync::P2P(gossip_sync) => Some(gossip_sync.network_graph()),
168 GossipSync::Rapid(gossip_sync) => {
169 if gossip_sync.is_initial_sync_complete() {
170 Some(gossip_sync.network_graph())
175 GossipSync::None => None,
180 /// This is not exported to bindings users as the bindings concretize everything and have constructors for us
181 impl<P: Deref<Target = P2PGossipSync<G, U, L>>, G: Deref<Target = NetworkGraph<L>>, U: Deref, L: Deref>
182 GossipSync<P, &RapidGossipSync<G, L>, G, U, L>
184 U::Target: UtxoLookup,
187 /// Initializes a new [`GossipSync::P2P`] variant.
188 pub fn p2p(gossip_sync: P) -> Self {
189 GossipSync::P2P(gossip_sync)
193 /// This is not exported to bindings users as the bindings concretize everything and have constructors for us
194 impl<'a, R: Deref<Target = RapidGossipSync<G, L>>, G: Deref<Target = NetworkGraph<L>>, L: Deref>
196 &P2PGossipSync<G, &'a (dyn UtxoLookup + Send + Sync), L>,
199 &'a (dyn UtxoLookup + Send + Sync),
205 /// Initializes a new [`GossipSync::Rapid`] variant.
206 pub fn rapid(gossip_sync: R) -> Self {
207 GossipSync::Rapid(gossip_sync)
211 /// This is not exported to bindings users as the bindings concretize everything and have constructors for us
214 &P2PGossipSync<&'a NetworkGraph<L>, &'a (dyn UtxoLookup + Send + Sync), L>,
215 &RapidGossipSync<&'a NetworkGraph<L>, L>,
217 &'a (dyn UtxoLookup + Send + Sync),
223 /// Initializes a new [`GossipSync::None`] variant.
224 pub fn none() -> Self {
229 fn handle_network_graph_update<L: Deref>(
230 network_graph: &NetworkGraph<L>, event: &Event
231 ) where L::Target: Logger {
232 if let Event::PaymentPathFailed {
233 failure: PathFailure::OnPath { network_update: Some(ref upd) }, .. } = event
235 network_graph.handle_network_update(upd);
239 /// Updates scorer based on event and returns whether an update occurred so we can decide whether
241 fn update_scorer<'a, S: 'static + Deref<Target = SC> + Send + Sync, SC: 'a + WriteableScore<'a>>(
242 scorer: &'a S, event: &Event
245 Event::PaymentPathFailed { ref path, short_channel_id: Some(scid), .. } => {
246 let mut score = scorer.write_lock();
247 score.payment_path_failed(path, *scid);
249 Event::PaymentPathFailed { ref path, payment_failed_permanently: true, .. } => {
250 // Reached if the destination explicitly failed it back. We treat this as a successful probe
251 // because the payment made it all the way to the destination with sufficient liquidity.
252 let mut score = scorer.write_lock();
253 score.probe_successful(path);
255 Event::PaymentPathSuccessful { path, .. } => {
256 let mut score = scorer.write_lock();
257 score.payment_path_successful(path);
259 Event::ProbeSuccessful { path, .. } => {
260 let mut score = scorer.write_lock();
261 score.probe_successful(path);
263 Event::ProbeFailed { path, short_channel_id: Some(scid), .. } => {
264 let mut score = scorer.write_lock();
265 score.probe_failed(path, *scid);
272 macro_rules! define_run_body {
273 ($persister: ident, $chain_monitor: ident, $process_chain_monitor_events: expr,
274 $channel_manager: ident, $process_channel_manager_events: expr,
275 $gossip_sync: ident, $peer_manager: ident, $logger: ident, $scorer: ident,
276 $loop_exit_check: expr, $await: expr, $get_timer: expr, $timer_elapsed: expr,
277 $check_slow_await: expr)
279 log_trace!($logger, "Calling ChannelManager's timer_tick_occurred on startup");
280 $channel_manager.timer_tick_occurred();
281 log_trace!($logger, "Rebroadcasting monitor's pending claims on startup");
282 $chain_monitor.rebroadcast_pending_claims();
284 let mut last_freshness_call = $get_timer(FRESHNESS_TIMER);
285 let mut last_ping_call = $get_timer(PING_TIMER);
286 let mut last_prune_call = $get_timer(FIRST_NETWORK_PRUNE_TIMER);
287 let mut last_scorer_persist_call = $get_timer(SCORER_PERSIST_TIMER);
288 let mut last_rebroadcast_call = $get_timer(REBROADCAST_TIMER);
289 let mut have_pruned = false;
292 $process_channel_manager_events;
293 $process_chain_monitor_events;
295 // Note that the PeerManager::process_events may block on ChannelManager's locks,
296 // hence it comes last here. When the ChannelManager finishes whatever it's doing,
297 // we want to ensure we get into `persist_manager` as quickly as we can, especially
298 // without running the normal event processing above and handing events to users.
300 // Specifically, on an *extremely* slow machine, we may see ChannelManager start
301 // processing a message effectively at any point during this loop. In order to
302 // minimize the time between such processing completing and persisting the updated
303 // ChannelManager, we want to minimize methods blocking on a ChannelManager
304 // generally, and as a fallback place such blocking only immediately before
306 $peer_manager.as_ref().process_events();
308 // Exit the loop if the background processor was requested to stop.
309 if $loop_exit_check {
310 log_trace!($logger, "Terminating background processor.");
314 // We wait up to 100ms, but track how long it takes to detect being put to sleep,
315 // see `await_start`'s use below.
316 let mut await_start = None;
317 if $check_slow_await { await_start = Some($get_timer(1)); }
319 let await_slow = if $check_slow_await { $timer_elapsed(&mut await_start.unwrap(), 1) } else { false };
321 // Exit the loop if the background processor was requested to stop.
322 if $loop_exit_check {
323 log_trace!($logger, "Terminating background processor.");
327 if $channel_manager.get_and_clear_needs_persistence() {
328 log_trace!($logger, "Persisting ChannelManager...");
329 $persister.persist_manager(&*$channel_manager)?;
330 log_trace!($logger, "Done persisting ChannelManager.");
332 if $timer_elapsed(&mut last_freshness_call, FRESHNESS_TIMER) {
333 log_trace!($logger, "Calling ChannelManager's timer_tick_occurred");
334 $channel_manager.timer_tick_occurred();
335 last_freshness_call = $get_timer(FRESHNESS_TIMER);
338 // On various platforms, we may be starved of CPU cycles for several reasons.
339 // E.g. on iOS, if we've been in the background, we will be entirely paused.
340 // Similarly, if we're on a desktop platform and the device has been asleep, we
341 // may not get any cycles.
342 // We detect this by checking if our max-100ms-sleep, above, ran longer than a
343 // full second, at which point we assume sockets may have been killed (they
344 // appear to be at least on some platforms, even if it has only been a second).
345 // Note that we have to take care to not get here just because user event
346 // processing was slow at the top of the loop. For example, the sample client
347 // may call Bitcoin Core RPCs during event handling, which very often takes
348 // more than a handful of seconds to complete, and shouldn't disconnect all our
350 log_trace!($logger, "100ms sleep took more than a second, disconnecting peers.");
351 $peer_manager.as_ref().disconnect_all_peers();
352 last_ping_call = $get_timer(PING_TIMER);
353 } else if $timer_elapsed(&mut last_ping_call, PING_TIMER) {
354 log_trace!($logger, "Calling PeerManager's timer_tick_occurred");
355 $peer_manager.as_ref().timer_tick_occurred();
356 last_ping_call = $get_timer(PING_TIMER);
359 // Note that we want to run a graph prune once not long after startup before
360 // falling back to our usual hourly prunes. This avoids short-lived clients never
361 // pruning their network graph. We run once 60 seconds after startup before
362 // continuing our normal cadence. For RGS, since 60 seconds is likely too long,
363 // we prune after an initial sync completes.
364 let prune_timer = if have_pruned { NETWORK_PRUNE_TIMER } else { FIRST_NETWORK_PRUNE_TIMER };
365 let prune_timer_elapsed = $timer_elapsed(&mut last_prune_call, prune_timer);
366 let should_prune = match $gossip_sync {
367 GossipSync::Rapid(_) => !have_pruned || prune_timer_elapsed,
368 _ => prune_timer_elapsed,
371 // The network graph must not be pruned while rapid sync completion is pending
372 if let Some(network_graph) = $gossip_sync.prunable_network_graph() {
373 #[cfg(feature = "std")] {
374 log_trace!($logger, "Pruning and persisting network graph.");
375 network_graph.remove_stale_channels_and_tracking();
377 #[cfg(not(feature = "std"))] {
378 log_warn!($logger, "Not pruning network graph, consider enabling `std` or doing so manually with remove_stale_channels_and_tracking_with_time.");
379 log_trace!($logger, "Persisting network graph.");
382 if let Err(e) = $persister.persist_graph(network_graph) {
383 log_error!($logger, "Error: Failed to persist network graph, check your disk and permissions {}", e)
388 let prune_timer = if have_pruned { NETWORK_PRUNE_TIMER } else { FIRST_NETWORK_PRUNE_TIMER };
389 last_prune_call = $get_timer(prune_timer);
392 if $timer_elapsed(&mut last_scorer_persist_call, SCORER_PERSIST_TIMER) {
393 if let Some(ref scorer) = $scorer {
394 log_trace!($logger, "Persisting scorer");
395 if let Err(e) = $persister.persist_scorer(&scorer) {
396 log_error!($logger, "Error: Failed to persist scorer, check your disk and permissions {}", e)
399 last_scorer_persist_call = $get_timer(SCORER_PERSIST_TIMER);
402 if $timer_elapsed(&mut last_rebroadcast_call, REBROADCAST_TIMER) {
403 log_trace!($logger, "Rebroadcasting monitor's pending claims");
404 $chain_monitor.rebroadcast_pending_claims();
405 last_rebroadcast_call = $get_timer(REBROADCAST_TIMER);
409 // After we exit, ensure we persist the ChannelManager one final time - this avoids
410 // some races where users quit while channel updates were in-flight, with
411 // ChannelMonitor update(s) persisted without a corresponding ChannelManager update.
412 $persister.persist_manager(&*$channel_manager)?;
414 // Persist Scorer on exit
415 if let Some(ref scorer) = $scorer {
416 $persister.persist_scorer(&scorer)?;
419 // Persist NetworkGraph on exit
420 if let Some(network_graph) = $gossip_sync.network_graph() {
421 $persister.persist_graph(network_graph)?;
428 #[cfg(feature = "futures")]
429 pub(crate) mod futures_util {
430 use core::future::Future;
431 use core::task::{Poll, Waker, RawWaker, RawWakerVTable};
433 use core::marker::Unpin;
434 pub(crate) struct Selector<
435 A: Future<Output=()> + Unpin, B: Future<Output=()> + Unpin, C: Future<Output=bool> + Unpin
441 pub(crate) enum SelectorOutput {
446 A: Future<Output=()> + Unpin, B: Future<Output=()> + Unpin, C: Future<Output=bool> + Unpin
447 > Future for Selector<A, B, C> {
448 type Output = SelectorOutput;
449 fn poll(mut self: Pin<&mut Self>, ctx: &mut core::task::Context<'_>) -> Poll<SelectorOutput> {
450 match Pin::new(&mut self.a).poll(ctx) {
451 Poll::Ready(()) => { return Poll::Ready(SelectorOutput::A); },
454 match Pin::new(&mut self.b).poll(ctx) {
455 Poll::Ready(()) => { return Poll::Ready(SelectorOutput::B); },
458 match Pin::new(&mut self.c).poll(ctx) {
459 Poll::Ready(res) => { return Poll::Ready(SelectorOutput::C(res)); },
466 // If we want to poll a future without an async context to figure out if it has completed or
467 // not without awaiting, we need a Waker, which needs a vtable...we fill it with dummy values
468 // but sadly there's a good bit of boilerplate here.
469 fn dummy_waker_clone(_: *const ()) -> RawWaker { RawWaker::new(core::ptr::null(), &DUMMY_WAKER_VTABLE) }
470 fn dummy_waker_action(_: *const ()) { }
472 const DUMMY_WAKER_VTABLE: RawWakerVTable = RawWakerVTable::new(
473 dummy_waker_clone, dummy_waker_action, dummy_waker_action, dummy_waker_action);
474 pub(crate) fn dummy_waker() -> Waker { unsafe { Waker::from_raw(RawWaker::new(core::ptr::null(), &DUMMY_WAKER_VTABLE)) } }
476 #[cfg(feature = "futures")]
477 use futures_util::{Selector, SelectorOutput, dummy_waker};
478 #[cfg(feature = "futures")]
481 /// Processes background events in a future.
483 /// `sleeper` should return a future which completes in the given amount of time and returns a
484 /// boolean indicating whether the background processing should exit. Once `sleeper` returns a
485 /// future which outputs `true`, the loop will exit and this function's future will complete.
486 /// The `sleeper` future is free to return early after it has triggered the exit condition.
488 /// See [`BackgroundProcessor::start`] for information on which actions this handles.
490 /// Requires the `futures` feature. Note that while this method is available without the `std`
491 /// feature, doing so will skip calling [`NetworkGraph::remove_stale_channels_and_tracking`],
492 /// you should call [`NetworkGraph::remove_stale_channels_and_tracking_with_time`] regularly
493 /// manually instead.
495 /// The `mobile_interruptable_platform` flag should be set if we're currently running on a
496 /// mobile device, where we may need to check for interruption of the application regularly. If you
497 /// are unsure, you should set the flag, as the performance impact of it is minimal unless there
498 /// are hundreds or thousands of simultaneous process calls running.
500 /// For example, in order to process background events in a [Tokio](https://tokio.rs/) task, you
501 /// could setup `process_events_async` like this:
503 /// # use lightning::io;
504 /// # use std::sync::{Arc, Mutex};
505 /// # use std::sync::atomic::{AtomicBool, Ordering};
506 /// # use lightning_background_processor::{process_events_async, GossipSync};
507 /// # struct MyStore {}
508 /// # impl lightning::util::persist::KVStore for MyStore {
509 /// # fn read(&self, namespace: &str, sub_namespace: &str, key: &str) -> io::Result<Vec<u8>> { Ok(Vec::new()) }
510 /// # fn write(&self, namespace: &str, sub_namespace: &str, key: &str, buf: &[u8]) -> io::Result<()> { Ok(()) }
511 /// # fn remove(&self, namespace: &str, sub_namespace: &str, key: &str, lazy: bool) -> io::Result<()> { Ok(()) }
512 /// # fn list(&self, namespace: &str, sub_namespace: &str) -> io::Result<Vec<String>> { Ok(Vec::new()) }
514 /// # struct MyEventHandler {}
515 /// # impl MyEventHandler {
516 /// # async fn handle_event(&self, _: lightning::events::Event) {}
518 /// # #[derive(Eq, PartialEq, Clone, Hash)]
519 /// # struct MySocketDescriptor {}
520 /// # impl lightning::ln::peer_handler::SocketDescriptor for MySocketDescriptor {
521 /// # fn send_data(&mut self, _data: &[u8], _resume_read: bool) -> usize { 0 }
522 /// # fn disconnect_socket(&mut self) {}
524 /// # type MyBroadcaster = dyn lightning::chain::chaininterface::BroadcasterInterface + Send + Sync;
525 /// # type MyFeeEstimator = dyn lightning::chain::chaininterface::FeeEstimator + Send + Sync;
526 /// # type MyNodeSigner = dyn lightning::sign::NodeSigner + Send + Sync;
527 /// # type MyUtxoLookup = dyn lightning::routing::utxo::UtxoLookup + Send + Sync;
528 /// # type MyFilter = dyn lightning::chain::Filter + Send + Sync;
529 /// # type MyLogger = dyn lightning::util::logger::Logger + Send + Sync;
530 /// # type MyChainMonitor = lightning::chain::chainmonitor::ChainMonitor<lightning::sign::InMemorySigner, Arc<MyFilter>, Arc<MyBroadcaster>, Arc<MyFeeEstimator>, Arc<MyLogger>, Arc<MyStore>>;
531 /// # type MyPeerManager = lightning::ln::peer_handler::SimpleArcPeerManager<MySocketDescriptor, MyChainMonitor, MyBroadcaster, MyFeeEstimator, MyUtxoLookup, MyLogger>;
532 /// # type MyNetworkGraph = lightning::routing::gossip::NetworkGraph<Arc<MyLogger>>;
533 /// # type MyGossipSync = lightning::routing::gossip::P2PGossipSync<Arc<MyNetworkGraph>, Arc<MyUtxoLookup>, Arc<MyLogger>>;
534 /// # type MyChannelManager = lightning::ln::channelmanager::SimpleArcChannelManager<MyChainMonitor, MyBroadcaster, MyFeeEstimator, MyLogger>;
535 /// # type MyScorer = Mutex<lightning::routing::scoring::ProbabilisticScorer<Arc<MyNetworkGraph>, Arc<MyLogger>>>;
537 /// # async fn setup_background_processing(my_persister: Arc<MyStore>, my_event_handler: Arc<MyEventHandler>, my_chain_monitor: Arc<MyChainMonitor>, my_channel_manager: Arc<MyChannelManager>, my_gossip_sync: Arc<MyGossipSync>, my_logger: Arc<MyLogger>, my_scorer: Arc<MyScorer>, my_peer_manager: Arc<MyPeerManager>) {
538 /// let background_persister = Arc::clone(&my_persister);
539 /// let background_event_handler = Arc::clone(&my_event_handler);
540 /// let background_chain_mon = Arc::clone(&my_chain_monitor);
541 /// let background_chan_man = Arc::clone(&my_channel_manager);
542 /// let background_gossip_sync = GossipSync::p2p(Arc::clone(&my_gossip_sync));
543 /// let background_peer_man = Arc::clone(&my_peer_manager);
544 /// let background_logger = Arc::clone(&my_logger);
545 /// let background_scorer = Arc::clone(&my_scorer);
547 /// // Setup the sleeper.
548 /// let (stop_sender, stop_receiver) = tokio::sync::watch::channel(());
550 /// let sleeper = move |d| {
551 /// let mut receiver = stop_receiver.clone();
552 /// Box::pin(async move {
554 /// _ = tokio::time::sleep(d) => false,
555 /// _ = receiver.changed() => true,
560 /// let mobile_interruptable_platform = false;
562 /// let handle = tokio::spawn(async move {
563 /// process_events_async(
564 /// background_persister,
565 /// |e| background_event_handler.handle_event(e),
566 /// background_chain_mon,
567 /// background_chan_man,
568 /// background_gossip_sync,
569 /// background_peer_man,
570 /// background_logger,
571 /// Some(background_scorer),
573 /// mobile_interruptable_platform,
576 /// .expect("Failed to process events");
579 /// // Stop the background processing.
580 /// stop_sender.send(()).unwrap();
581 /// handle.await.unwrap();
584 #[cfg(feature = "futures")]
585 pub async fn process_events_async<
587 UL: 'static + Deref + Send + Sync,
588 CF: 'static + Deref + Send + Sync,
589 CW: 'static + Deref + Send + Sync,
590 T: 'static + Deref + Send + Sync,
591 ES: 'static + Deref + Send + Sync,
592 NS: 'static + Deref + Send + Sync,
593 SP: 'static + Deref + Send + Sync,
594 F: 'static + Deref + Send + Sync,
595 R: 'static + Deref + Send + Sync,
596 G: 'static + Deref<Target = NetworkGraph<L>> + Send + Sync,
597 L: 'static + Deref + Send + Sync,
598 P: 'static + Deref + Send + Sync,
599 EventHandlerFuture: core::future::Future<Output = ()>,
600 EventHandler: Fn(Event) -> EventHandlerFuture,
601 PS: 'static + Deref + Send,
602 M: 'static + Deref<Target = ChainMonitor<<SP::Target as SignerProvider>::Signer, CF, T, F, L, P>> + Send + Sync,
603 CM: 'static + Deref<Target = ChannelManager<CW, T, ES, NS, SP, F, R, L>> + Send + Sync,
604 PGS: 'static + Deref<Target = P2PGossipSync<G, UL, L>> + Send + Sync,
605 RGS: 'static + Deref<Target = RapidGossipSync<G, L>> + Send,
606 APM: APeerManager + Send + Sync,
607 PM: 'static + Deref<Target = APM> + Send + Sync,
608 S: 'static + Deref<Target = SC> + Send + Sync,
609 SC: for<'b> WriteableScore<'b>,
610 SleepFuture: core::future::Future<Output = bool> + core::marker::Unpin,
611 Sleeper: Fn(Duration) -> SleepFuture
613 persister: PS, event_handler: EventHandler, chain_monitor: M, channel_manager: CM,
614 gossip_sync: GossipSync<PGS, RGS, G, UL, L>, peer_manager: PM, logger: L, scorer: Option<S>,
615 sleeper: Sleeper, mobile_interruptable_platform: bool,
616 ) -> Result<(), lightning::io::Error>
618 UL::Target: 'static + UtxoLookup,
619 CF::Target: 'static + chain::Filter,
620 CW::Target: 'static + chain::Watch<<SP::Target as SignerProvider>::Signer>,
621 T::Target: 'static + BroadcasterInterface,
622 ES::Target: 'static + EntropySource,
623 NS::Target: 'static + NodeSigner,
624 SP::Target: 'static + SignerProvider,
625 F::Target: 'static + FeeEstimator,
626 R::Target: 'static + Router,
627 L::Target: 'static + Logger,
628 P::Target: 'static + Persist<<SP::Target as SignerProvider>::Signer>,
629 PS::Target: 'static + Persister<'a, CW, T, ES, NS, SP, F, R, L, SC>,
631 let mut should_break = false;
632 let async_event_handler = |event| {
633 let network_graph = gossip_sync.network_graph();
634 let event_handler = &event_handler;
635 let scorer = &scorer;
636 let logger = &logger;
637 let persister = &persister;
639 if let Some(network_graph) = network_graph {
640 handle_network_graph_update(network_graph, &event)
642 if let Some(ref scorer) = scorer {
643 if update_scorer(scorer, &event) {
644 log_trace!(logger, "Persisting scorer after update");
645 if let Err(e) = persister.persist_scorer(&scorer) {
646 log_error!(logger, "Error: Failed to persist scorer, check your disk and permissions {}", e)
650 event_handler(event).await;
653 define_run_body!(persister,
654 chain_monitor, chain_monitor.process_pending_events_async(async_event_handler).await,
655 channel_manager, channel_manager.process_pending_events_async(async_event_handler).await,
656 gossip_sync, peer_manager, logger, scorer, should_break, {
658 a: channel_manager.get_event_or_persistence_needed_future(),
659 b: chain_monitor.get_update_future(),
660 c: sleeper(if mobile_interruptable_platform { Duration::from_millis(100) } else { Duration::from_secs(FASTEST_TIMER) }),
663 SelectorOutput::A|SelectorOutput::B => {},
664 SelectorOutput::C(exit) => {
668 }, |t| sleeper(Duration::from_secs(t)),
669 |fut: &mut SleepFuture, _| {
670 let mut waker = dummy_waker();
671 let mut ctx = task::Context::from_waker(&mut waker);
672 match core::pin::Pin::new(fut).poll(&mut ctx) {
673 task::Poll::Ready(exit) => { should_break = exit; true },
674 task::Poll::Pending => false,
676 }, mobile_interruptable_platform)
679 #[cfg(feature = "std")]
680 impl BackgroundProcessor {
681 /// Start a background thread that takes care of responsibilities enumerated in the [top-level
684 /// The thread runs indefinitely unless the object is dropped, [`stop`] is called, or
685 /// [`Persister::persist_manager`] returns an error. In case of an error, the error is retrieved by calling
686 /// either [`join`] or [`stop`].
688 /// # Data Persistence
690 /// [`Persister::persist_manager`] is responsible for writing out the [`ChannelManager`] to disk, and/or
691 /// uploading to one or more backup services. See [`ChannelManager::write`] for writing out a
692 /// [`ChannelManager`]. See the `lightning-persister` crate for LDK's
693 /// provided implementation.
695 /// [`Persister::persist_graph`] is responsible for writing out the [`NetworkGraph`] to disk, if
696 /// [`GossipSync`] is supplied. See [`NetworkGraph::write`] for writing out a [`NetworkGraph`].
697 /// See the `lightning-persister` crate for LDK's provided implementation.
699 /// Typically, users should either implement [`Persister::persist_manager`] to never return an
700 /// error or call [`join`] and handle any error that may arise. For the latter case,
701 /// `BackgroundProcessor` must be restarted by calling `start` again after handling the error.
705 /// `event_handler` is responsible for handling events that users should be notified of (e.g.,
706 /// payment failed). [`BackgroundProcessor`] may decorate the given [`EventHandler`] with common
707 /// functionality implemented by other handlers.
708 /// * [`P2PGossipSync`] if given will update the [`NetworkGraph`] based on payment failures.
710 /// # Rapid Gossip Sync
712 /// If rapid gossip sync is meant to run at startup, pass [`RapidGossipSync`] via `gossip_sync`
713 /// to indicate that the [`BackgroundProcessor`] should not prune the [`NetworkGraph`] instance
714 /// until the [`RapidGossipSync`] instance completes its first sync.
716 /// [top-level documentation]: BackgroundProcessor
717 /// [`join`]: Self::join
718 /// [`stop`]: Self::stop
719 /// [`ChannelManager`]: lightning::ln::channelmanager::ChannelManager
720 /// [`ChannelManager::write`]: lightning::ln::channelmanager::ChannelManager#impl-Writeable
721 /// [`Persister::persist_manager`]: lightning::util::persist::Persister::persist_manager
722 /// [`Persister::persist_graph`]: lightning::util::persist::Persister::persist_graph
723 /// [`NetworkGraph`]: lightning::routing::gossip::NetworkGraph
724 /// [`NetworkGraph::write`]: lightning::routing::gossip::NetworkGraph#impl-Writeable
727 UL: 'static + Deref + Send + Sync,
728 CF: 'static + Deref + Send + Sync,
729 CW: 'static + Deref + Send + Sync,
730 T: 'static + Deref + Send + Sync,
731 ES: 'static + Deref + Send + Sync,
732 NS: 'static + Deref + Send + Sync,
733 SP: 'static + Deref + Send + Sync,
734 F: 'static + Deref + Send + Sync,
735 R: 'static + Deref + Send + Sync,
736 G: 'static + Deref<Target = NetworkGraph<L>> + Send + Sync,
737 L: 'static + Deref + Send + Sync,
738 P: 'static + Deref + Send + Sync,
739 EH: 'static + EventHandler + Send,
740 PS: 'static + Deref + Send,
741 M: 'static + Deref<Target = ChainMonitor<<SP::Target as SignerProvider>::Signer, CF, T, F, L, P>> + Send + Sync,
742 CM: 'static + Deref<Target = ChannelManager<CW, T, ES, NS, SP, F, R, L>> + Send + Sync,
743 PGS: 'static + Deref<Target = P2PGossipSync<G, UL, L>> + Send + Sync,
744 RGS: 'static + Deref<Target = RapidGossipSync<G, L>> + Send,
745 APM: APeerManager + Send + Sync,
746 PM: 'static + Deref<Target = APM> + Send + Sync,
747 S: 'static + Deref<Target = SC> + Send + Sync,
748 SC: for <'b> WriteableScore<'b>,
750 persister: PS, event_handler: EH, chain_monitor: M, channel_manager: CM,
751 gossip_sync: GossipSync<PGS, RGS, G, UL, L>, peer_manager: PM, logger: L, scorer: Option<S>,
754 UL::Target: 'static + UtxoLookup,
755 CF::Target: 'static + chain::Filter,
756 CW::Target: 'static + chain::Watch<<SP::Target as SignerProvider>::Signer>,
757 T::Target: 'static + BroadcasterInterface,
758 ES::Target: 'static + EntropySource,
759 NS::Target: 'static + NodeSigner,
760 SP::Target: 'static + SignerProvider,
761 F::Target: 'static + FeeEstimator,
762 R::Target: 'static + Router,
763 L::Target: 'static + Logger,
764 P::Target: 'static + Persist<<SP::Target as SignerProvider>::Signer>,
765 PS::Target: 'static + Persister<'a, CW, T, ES, NS, SP, F, R, L, SC>,
767 let stop_thread = Arc::new(AtomicBool::new(false));
768 let stop_thread_clone = stop_thread.clone();
769 let handle = thread::spawn(move || -> Result<(), std::io::Error> {
770 let event_handler = |event| {
771 let network_graph = gossip_sync.network_graph();
772 if let Some(network_graph) = network_graph {
773 handle_network_graph_update(network_graph, &event)
775 if let Some(ref scorer) = scorer {
776 if update_scorer(scorer, &event) {
777 log_trace!(logger, "Persisting scorer after update");
778 if let Err(e) = persister.persist_scorer(&scorer) {
779 log_error!(logger, "Error: Failed to persist scorer, check your disk and permissions {}", e)
783 event_handler.handle_event(event);
785 define_run_body!(persister, chain_monitor, chain_monitor.process_pending_events(&event_handler),
786 channel_manager, channel_manager.process_pending_events(&event_handler),
787 gossip_sync, peer_manager, logger, scorer, stop_thread.load(Ordering::Acquire),
788 { Sleeper::from_two_futures(
789 channel_manager.get_event_or_persistence_needed_future(),
790 chain_monitor.get_update_future()
791 ).wait_timeout(Duration::from_millis(100)); },
792 |_| Instant::now(), |time: &Instant, dur| time.elapsed().as_secs() > dur, false)
794 Self { stop_thread: stop_thread_clone, thread_handle: Some(handle) }
797 /// Join `BackgroundProcessor`'s thread, returning any error that occurred while persisting
798 /// [`ChannelManager`].
802 /// This function panics if the background thread has panicked such as while persisting or
805 /// [`ChannelManager`]: lightning::ln::channelmanager::ChannelManager
806 pub fn join(mut self) -> Result<(), std::io::Error> {
807 assert!(self.thread_handle.is_some());
811 /// Stop `BackgroundProcessor`'s thread, returning any error that occurred while persisting
812 /// [`ChannelManager`].
816 /// This function panics if the background thread has panicked such as while persisting or
819 /// [`ChannelManager`]: lightning::ln::channelmanager::ChannelManager
820 pub fn stop(mut self) -> Result<(), std::io::Error> {
821 assert!(self.thread_handle.is_some());
822 self.stop_and_join_thread()
825 fn stop_and_join_thread(&mut self) -> Result<(), std::io::Error> {
826 self.stop_thread.store(true, Ordering::Release);
830 fn join_thread(&mut self) -> Result<(), std::io::Error> {
831 match self.thread_handle.take() {
832 Some(handle) => handle.join().unwrap(),
838 #[cfg(feature = "std")]
839 impl Drop for BackgroundProcessor {
841 self.stop_and_join_thread().unwrap();
845 #[cfg(all(feature = "std", test))]
847 use bitcoin::blockdata::constants::{genesis_block, ChainHash};
848 use bitcoin::blockdata::locktime::PackedLockTime;
849 use bitcoin::blockdata::transaction::{Transaction, TxOut};
850 use bitcoin::network::constants::Network;
851 use bitcoin::secp256k1::{SecretKey, PublicKey, Secp256k1};
852 use lightning::chain::{BestBlock, Confirm, chainmonitor};
853 use lightning::chain::channelmonitor::ANTI_REORG_DELAY;
854 use lightning::sign::{InMemorySigner, KeysManager};
855 use lightning::chain::transaction::OutPoint;
856 use lightning::events::{Event, PathFailure, MessageSendEventsProvider, MessageSendEvent};
857 use lightning::{get_event_msg, get_event};
858 use lightning::ln::PaymentHash;
859 use lightning::ln::channelmanager;
860 use lightning::ln::channelmanager::{BREAKDOWN_TIMEOUT, ChainParameters, MIN_CLTV_EXPIRY_DELTA, PaymentId};
861 use lightning::ln::features::{ChannelFeatures, NodeFeatures};
862 use lightning::ln::functional_test_utils::*;
863 use lightning::ln::msgs::{ChannelMessageHandler, Init};
864 use lightning::ln::peer_handler::{PeerManager, MessageHandler, SocketDescriptor, IgnoringMessageHandler};
865 use lightning::routing::gossip::{NetworkGraph, NodeId, P2PGossipSync};
866 use lightning::routing::router::{DefaultRouter, Path, RouteHop};
867 use lightning::routing::scoring::{ChannelUsage, ScoreUpdate, ScoreLookUp};
868 use lightning::util::config::UserConfig;
869 use lightning::util::ser::Writeable;
870 use lightning::util::test_utils;
871 use lightning::util::persist::{KVStore, CHANNEL_MANAGER_PERSISTENCE_NAMESPACE, CHANNEL_MANAGER_PERSISTENCE_SUB_NAMESPACE, CHANNEL_MANAGER_PERSISTENCE_KEY, NETWORK_GRAPH_PERSISTENCE_NAMESPACE, NETWORK_GRAPH_PERSISTENCE_SUB_NAMESPACE, NETWORK_GRAPH_PERSISTENCE_KEY, SCORER_PERSISTENCE_NAMESPACE, SCORER_PERSISTENCE_SUB_NAMESPACE, SCORER_PERSISTENCE_KEY};
872 use lightning_persister::fs_store::FilesystemStore;
873 use std::collections::VecDeque;
875 use std::path::PathBuf;
876 use std::sync::{Arc, Mutex};
877 use std::sync::mpsc::SyncSender;
878 use std::time::Duration;
879 use lightning_rapid_gossip_sync::RapidGossipSync;
880 use super::{BackgroundProcessor, GossipSync, FRESHNESS_TIMER};
882 const EVENT_DEADLINE: u64 = 5 * FRESHNESS_TIMER;
884 #[derive(Clone, Hash, PartialEq, Eq)]
885 struct TestDescriptor{}
886 impl SocketDescriptor for TestDescriptor {
887 fn send_data(&mut self, _data: &[u8], _resume_read: bool) -> usize {
891 fn disconnect_socket(&mut self) {}
894 type ChannelManager =
895 channelmanager::ChannelManager<
897 Arc<test_utils::TestBroadcaster>,
901 Arc<test_utils::TestFeeEstimator>,
903 Arc<NetworkGraph<Arc<test_utils::TestLogger>>>,
904 Arc<test_utils::TestLogger>,
905 Arc<Mutex<TestScorer>>,
909 Arc<test_utils::TestLogger>>;
911 type ChainMonitor = chainmonitor::ChainMonitor<InMemorySigner, Arc<test_utils::TestChainSource>, Arc<test_utils::TestBroadcaster>, Arc<test_utils::TestFeeEstimator>, Arc<test_utils::TestLogger>, Arc<FilesystemStore>>;
913 type PGS = Arc<P2PGossipSync<Arc<NetworkGraph<Arc<test_utils::TestLogger>>>, Arc<test_utils::TestChainSource>, Arc<test_utils::TestLogger>>>;
914 type RGS = Arc<RapidGossipSync<Arc<NetworkGraph<Arc<test_utils::TestLogger>>>, Arc<test_utils::TestLogger>>>;
917 node: Arc<ChannelManager>,
918 p2p_gossip_sync: PGS,
919 rapid_gossip_sync: RGS,
920 peer_manager: Arc<PeerManager<TestDescriptor, Arc<test_utils::TestChannelMessageHandler>, Arc<test_utils::TestRoutingMessageHandler>, IgnoringMessageHandler, Arc<test_utils::TestLogger>, IgnoringMessageHandler, Arc<KeysManager>>>,
921 chain_monitor: Arc<ChainMonitor>,
922 kv_store: Arc<FilesystemStore>,
923 tx_broadcaster: Arc<test_utils::TestBroadcaster>,
924 network_graph: Arc<NetworkGraph<Arc<test_utils::TestLogger>>>,
925 logger: Arc<test_utils::TestLogger>,
926 best_block: BestBlock,
927 scorer: Arc<Mutex<TestScorer>>,
931 fn p2p_gossip_sync(&self) -> GossipSync<PGS, RGS, Arc<NetworkGraph<Arc<test_utils::TestLogger>>>, Arc<test_utils::TestChainSource>, Arc<test_utils::TestLogger>> {
932 GossipSync::P2P(self.p2p_gossip_sync.clone())
935 fn rapid_gossip_sync(&self) -> GossipSync<PGS, RGS, Arc<NetworkGraph<Arc<test_utils::TestLogger>>>, Arc<test_utils::TestChainSource>, Arc<test_utils::TestLogger>> {
936 GossipSync::Rapid(self.rapid_gossip_sync.clone())
939 fn no_gossip_sync(&self) -> GossipSync<PGS, RGS, Arc<NetworkGraph<Arc<test_utils::TestLogger>>>, Arc<test_utils::TestChainSource>, Arc<test_utils::TestLogger>> {
946 let data_dir = self.kv_store.get_data_dir();
947 match fs::remove_dir_all(data_dir.clone()) {
948 Err(e) => println!("Failed to remove test store directory {}: {}", data_dir.display(), e),
955 graph_error: Option<(std::io::ErrorKind, &'static str)>,
956 graph_persistence_notifier: Option<SyncSender<()>>,
957 manager_error: Option<(std::io::ErrorKind, &'static str)>,
958 scorer_error: Option<(std::io::ErrorKind, &'static str)>,
959 kv_store: FilesystemStore,
963 fn new(data_dir: PathBuf) -> Self {
964 let kv_store = FilesystemStore::new(data_dir);
965 Self { graph_error: None, graph_persistence_notifier: None, manager_error: None, scorer_error: None, kv_store }
968 fn with_graph_error(self, error: std::io::ErrorKind, message: &'static str) -> Self {
969 Self { graph_error: Some((error, message)), ..self }
972 fn with_graph_persistence_notifier(self, sender: SyncSender<()>) -> Self {
973 Self { graph_persistence_notifier: Some(sender), ..self }
976 fn with_manager_error(self, error: std::io::ErrorKind, message: &'static str) -> Self {
977 Self { manager_error: Some((error, message)), ..self }
980 fn with_scorer_error(self, error: std::io::ErrorKind, message: &'static str) -> Self {
981 Self { scorer_error: Some((error, message)), ..self }
985 impl KVStore for Persister {
986 fn read(&self, namespace: &str, sub_namespace: &str, key: &str) -> lightning::io::Result<Vec<u8>> {
987 self.kv_store.read(namespace, sub_namespace, key)
990 fn write(&self, namespace: &str, sub_namespace: &str, key: &str, buf: &[u8]) -> lightning::io::Result<()> {
991 if namespace == CHANNEL_MANAGER_PERSISTENCE_NAMESPACE &&
992 sub_namespace == CHANNEL_MANAGER_PERSISTENCE_SUB_NAMESPACE &&
993 key == CHANNEL_MANAGER_PERSISTENCE_KEY
995 if let Some((error, message)) = self.manager_error {
996 return Err(std::io::Error::new(error, message))
1000 if namespace == NETWORK_GRAPH_PERSISTENCE_NAMESPACE &&
1001 sub_namespace == NETWORK_GRAPH_PERSISTENCE_SUB_NAMESPACE &&
1002 key == NETWORK_GRAPH_PERSISTENCE_KEY
1004 if let Some(sender) = &self.graph_persistence_notifier {
1005 match sender.send(()) {
1007 Err(std::sync::mpsc::SendError(())) => println!("Persister failed to notify as receiver went away."),
1011 if let Some((error, message)) = self.graph_error {
1012 return Err(std::io::Error::new(error, message))
1016 if namespace == SCORER_PERSISTENCE_NAMESPACE &&
1017 sub_namespace == SCORER_PERSISTENCE_SUB_NAMESPACE &&
1018 key == SCORER_PERSISTENCE_KEY
1020 if let Some((error, message)) = self.scorer_error {
1021 return Err(std::io::Error::new(error, message))
1025 self.kv_store.write(namespace, sub_namespace, key, buf)
1028 fn remove(&self, namespace: &str, sub_namespace: &str, key: &str, lazy: bool) -> lightning::io::Result<()> {
1029 self.kv_store.remove(namespace, sub_namespace, key, lazy)
1032 fn list(&self, namespace: &str, sub_namespace: &str) -> lightning::io::Result<Vec<String>> {
1033 self.kv_store.list(namespace, sub_namespace)
1038 event_expectations: Option<VecDeque<TestResult>>,
1043 PaymentFailure { path: Path, short_channel_id: u64 },
1044 PaymentSuccess { path: Path },
1045 ProbeFailure { path: Path },
1046 ProbeSuccess { path: Path },
1051 Self { event_expectations: None }
1054 fn expect(&mut self, expectation: TestResult) {
1055 self.event_expectations.get_or_insert_with(VecDeque::new).push_back(expectation);
1059 impl lightning::util::ser::Writeable for TestScorer {
1060 fn write<W: lightning::util::ser::Writer>(&self, _: &mut W) -> Result<(), lightning::io::Error> { Ok(()) }
1063 impl ScoreLookUp for TestScorer {
1064 type ScoreParams = ();
1065 fn channel_penalty_msat(
1066 &self, _short_channel_id: u64, _source: &NodeId, _target: &NodeId, _usage: ChannelUsage, _score_params: &Self::ScoreParams
1067 ) -> u64 { unimplemented!(); }
1070 impl ScoreUpdate for TestScorer {
1071 fn payment_path_failed(&mut self, actual_path: &Path, actual_short_channel_id: u64) {
1072 if let Some(expectations) = &mut self.event_expectations {
1073 match expectations.pop_front().unwrap() {
1074 TestResult::PaymentFailure { path, short_channel_id } => {
1075 assert_eq!(actual_path, &path);
1076 assert_eq!(actual_short_channel_id, short_channel_id);
1078 TestResult::PaymentSuccess { path } => {
1079 panic!("Unexpected successful payment path: {:?}", path)
1081 TestResult::ProbeFailure { path } => {
1082 panic!("Unexpected probe failure: {:?}", path)
1084 TestResult::ProbeSuccess { path } => {
1085 panic!("Unexpected probe success: {:?}", path)
1091 fn payment_path_successful(&mut self, actual_path: &Path) {
1092 if let Some(expectations) = &mut self.event_expectations {
1093 match expectations.pop_front().unwrap() {
1094 TestResult::PaymentFailure { path, .. } => {
1095 panic!("Unexpected payment path failure: {:?}", path)
1097 TestResult::PaymentSuccess { path } => {
1098 assert_eq!(actual_path, &path);
1100 TestResult::ProbeFailure { path } => {
1101 panic!("Unexpected probe failure: {:?}", path)
1103 TestResult::ProbeSuccess { path } => {
1104 panic!("Unexpected probe success: {:?}", path)
1110 fn probe_failed(&mut self, actual_path: &Path, _: u64) {
1111 if let Some(expectations) = &mut self.event_expectations {
1112 match expectations.pop_front().unwrap() {
1113 TestResult::PaymentFailure { path, .. } => {
1114 panic!("Unexpected payment path failure: {:?}", path)
1116 TestResult::PaymentSuccess { path } => {
1117 panic!("Unexpected payment path success: {:?}", path)
1119 TestResult::ProbeFailure { path } => {
1120 assert_eq!(actual_path, &path);
1122 TestResult::ProbeSuccess { path } => {
1123 panic!("Unexpected probe success: {:?}", path)
1128 fn probe_successful(&mut self, actual_path: &Path) {
1129 if let Some(expectations) = &mut self.event_expectations {
1130 match expectations.pop_front().unwrap() {
1131 TestResult::PaymentFailure { path, .. } => {
1132 panic!("Unexpected payment path failure: {:?}", path)
1134 TestResult::PaymentSuccess { path } => {
1135 panic!("Unexpected payment path success: {:?}", path)
1137 TestResult::ProbeFailure { path } => {
1138 panic!("Unexpected probe failure: {:?}", path)
1140 TestResult::ProbeSuccess { path } => {
1141 assert_eq!(actual_path, &path);
1148 impl Drop for TestScorer {
1149 fn drop(&mut self) {
1150 if std::thread::panicking() {
1154 if let Some(event_expectations) = &self.event_expectations {
1155 if !event_expectations.is_empty() {
1156 panic!("Unsatisfied event expectations: {:?}", event_expectations);
1162 fn get_full_filepath(filepath: String, filename: String) -> String {
1163 let mut path = PathBuf::from(filepath);
1164 path.push(filename);
1165 path.to_str().unwrap().to_string()
1168 fn create_nodes(num_nodes: usize, persist_dir: &str) -> (String, Vec<Node>) {
1169 let persist_temp_path = env::temp_dir().join(persist_dir);
1170 let persist_dir = persist_temp_path.to_string_lossy().to_string();
1171 let network = Network::Bitcoin;
1172 let mut nodes = Vec::new();
1173 for i in 0..num_nodes {
1174 let tx_broadcaster = Arc::new(test_utils::TestBroadcaster::new(network));
1175 let fee_estimator = Arc::new(test_utils::TestFeeEstimator { sat_per_kw: Mutex::new(253) });
1176 let logger = Arc::new(test_utils::TestLogger::with_id(format!("node {}", i)));
1177 let genesis_block = genesis_block(network);
1178 let network_graph = Arc::new(NetworkGraph::new(network, logger.clone()));
1179 let scorer = Arc::new(Mutex::new(TestScorer::new()));
1180 let seed = [i as u8; 32];
1181 let router = Arc::new(DefaultRouter::new(network_graph.clone(), logger.clone(), seed, scorer.clone(), ()));
1182 let chain_source = Arc::new(test_utils::TestChainSource::new(Network::Bitcoin));
1183 let kv_store = Arc::new(FilesystemStore::new(format!("{}_persister_{}", &persist_dir, i).into()));
1184 let now = Duration::from_secs(genesis_block.header.time as u64);
1185 let keys_manager = Arc::new(KeysManager::new(&seed, now.as_secs(), now.subsec_nanos()));
1186 let chain_monitor = Arc::new(chainmonitor::ChainMonitor::new(Some(chain_source.clone()), tx_broadcaster.clone(), logger.clone(), fee_estimator.clone(), kv_store.clone()));
1187 let best_block = BestBlock::from_network(network);
1188 let params = ChainParameters { network, best_block };
1189 let manager = Arc::new(ChannelManager::new(fee_estimator.clone(), chain_monitor.clone(), tx_broadcaster.clone(), router.clone(), logger.clone(), keys_manager.clone(), keys_manager.clone(), keys_manager.clone(), UserConfig::default(), params, genesis_block.header.time));
1190 let p2p_gossip_sync = Arc::new(P2PGossipSync::new(network_graph.clone(), Some(chain_source.clone()), logger.clone()));
1191 let rapid_gossip_sync = Arc::new(RapidGossipSync::new(network_graph.clone(), logger.clone()));
1192 let msg_handler = MessageHandler {
1193 chan_handler: Arc::new(test_utils::TestChannelMessageHandler::new(ChainHash::using_genesis_block(Network::Testnet))),
1194 route_handler: Arc::new(test_utils::TestRoutingMessageHandler::new()),
1195 onion_message_handler: IgnoringMessageHandler{}, custom_message_handler: IgnoringMessageHandler{}
1197 let peer_manager = Arc::new(PeerManager::new(msg_handler, 0, &seed, logger.clone(), keys_manager.clone()));
1198 let node = Node { node: manager, p2p_gossip_sync, rapid_gossip_sync, peer_manager, chain_monitor, kv_store, tx_broadcaster, network_graph, logger, best_block, scorer };
1202 for i in 0..num_nodes {
1203 for j in (i+1)..num_nodes {
1204 nodes[i].node.peer_connected(&nodes[j].node.get_our_node_id(), &Init {
1205 features: nodes[j].node.init_features(), networks: None, remote_network_address: None
1207 nodes[j].node.peer_connected(&nodes[i].node.get_our_node_id(), &Init {
1208 features: nodes[i].node.init_features(), networks: None, remote_network_address: None
1213 (persist_dir, nodes)
1216 macro_rules! open_channel {
1217 ($node_a: expr, $node_b: expr, $channel_value: expr) => {{
1218 begin_open_channel!($node_a, $node_b, $channel_value);
1219 let events = $node_a.node.get_and_clear_pending_events();
1220 assert_eq!(events.len(), 1);
1221 let (temporary_channel_id, tx) = handle_funding_generation_ready!(events[0], $channel_value);
1222 $node_a.node.funding_transaction_generated(&temporary_channel_id, &$node_b.node.get_our_node_id(), tx.clone()).unwrap();
1223 $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()));
1224 get_event!($node_b, Event::ChannelPending);
1225 $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()));
1226 get_event!($node_a, Event::ChannelPending);
1231 macro_rules! begin_open_channel {
1232 ($node_a: expr, $node_b: expr, $channel_value: expr) => {{
1233 $node_a.node.create_channel($node_b.node.get_our_node_id(), $channel_value, 100, 42, None).unwrap();
1234 $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()));
1235 $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()));
1239 macro_rules! handle_funding_generation_ready {
1240 ($event: expr, $channel_value: expr) => {{
1242 Event::FundingGenerationReady { temporary_channel_id, channel_value_satoshis, ref output_script, user_channel_id, .. } => {
1243 assert_eq!(channel_value_satoshis, $channel_value);
1244 assert_eq!(user_channel_id, 42);
1246 let tx = Transaction { version: 1 as i32, lock_time: PackedLockTime(0), input: Vec::new(), output: vec![TxOut {
1247 value: channel_value_satoshis, script_pubkey: output_script.clone(),
1249 (temporary_channel_id, tx)
1251 _ => panic!("Unexpected event"),
1256 fn confirm_transaction_depth(node: &mut Node, tx: &Transaction, depth: u32) {
1257 for i in 1..=depth {
1258 let prev_blockhash = node.best_block.block_hash();
1259 let height = node.best_block.height() + 1;
1260 let header = create_dummy_header(prev_blockhash, height);
1261 let txdata = vec![(0, tx)];
1262 node.best_block = BestBlock::new(header.block_hash(), height);
1265 node.node.transactions_confirmed(&header, &txdata, height);
1266 node.chain_monitor.transactions_confirmed(&header, &txdata, height);
1268 x if x == depth => {
1269 node.node.best_block_updated(&header, height);
1270 node.chain_monitor.best_block_updated(&header, height);
1276 fn confirm_transaction(node: &mut Node, tx: &Transaction) {
1277 confirm_transaction_depth(node, tx, ANTI_REORG_DELAY);
1281 fn test_background_processor() {
1282 // Test that when a new channel is created, the ChannelManager needs to be re-persisted with
1283 // updates. Also test that when new updates are available, the manager signals that it needs
1284 // re-persistence and is successfully re-persisted.
1285 let (persist_dir, nodes) = create_nodes(2, "test_background_processor");
1287 // Go through the channel creation process so that each node has something to persist. Since
1288 // open_channel consumes events, it must complete before starting BackgroundProcessor to
1289 // avoid a race with processing events.
1290 let tx = open_channel!(nodes[0], nodes[1], 100000);
1292 // Initiate the background processors to watch each node.
1293 let data_dir = nodes[0].kv_store.get_data_dir();
1294 let persister = Arc::new(Persister::new(data_dir));
1295 let event_handler = |_: _| {};
1296 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()));
1298 macro_rules! check_persisted_data {
1299 ($node: expr, $filepath: expr) => {
1300 let mut expected_bytes = Vec::new();
1302 expected_bytes.clear();
1303 match $node.write(&mut expected_bytes) {
1305 match std::fs::read($filepath) {
1307 if bytes == expected_bytes {
1316 Err(e) => panic!("Unexpected error: {}", e)
1322 // Check that the initial channel manager data is persisted as expected.
1323 let filepath = get_full_filepath(format!("{}_persister_0", &persist_dir), "manager".to_string());
1324 check_persisted_data!(nodes[0].node, filepath.clone());
1327 if !nodes[0].node.get_event_or_persist_condvar_value() { break }
1330 // Force-close the channel.
1331 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();
1333 // Check that the force-close updates are persisted.
1334 check_persisted_data!(nodes[0].node, filepath.clone());
1336 if !nodes[0].node.get_event_or_persist_condvar_value() { break }
1339 // Check network graph is persisted
1340 let filepath = get_full_filepath(format!("{}_persister_0", &persist_dir), "network_graph".to_string());
1341 check_persisted_data!(nodes[0].network_graph, filepath.clone());
1343 // Check scorer is persisted
1344 let filepath = get_full_filepath(format!("{}_persister_0", &persist_dir), "scorer".to_string());
1345 check_persisted_data!(nodes[0].scorer, filepath.clone());
1347 if !std::thread::panicking() {
1348 bg_processor.stop().unwrap();
1353 fn test_timer_tick_called() {
1354 // Test that `ChannelManager::timer_tick_occurred` is called every `FRESHNESS_TIMER`,
1355 // `ChainMonitor::rebroadcast_pending_claims` is called every `REBROADCAST_TIMER`, and
1356 // `PeerManager::timer_tick_occurred` every `PING_TIMER`.
1357 let (_, nodes) = create_nodes(1, "test_timer_tick_called");
1358 let data_dir = nodes[0].kv_store.get_data_dir();
1359 let persister = Arc::new(Persister::new(data_dir));
1360 let event_handler = |_: _| {};
1361 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()));
1363 let log_entries = nodes[0].logger.lines.lock().unwrap();
1364 let desired_log_1 = "Calling ChannelManager's timer_tick_occurred".to_string();
1365 let desired_log_2 = "Calling PeerManager's timer_tick_occurred".to_string();
1366 let desired_log_3 = "Rebroadcasting monitor's pending claims".to_string();
1367 if log_entries.get(&("lightning_background_processor".to_string(), desired_log_1)).is_some() &&
1368 log_entries.get(&("lightning_background_processor".to_string(), desired_log_2)).is_some() &&
1369 log_entries.get(&("lightning_background_processor".to_string(), desired_log_3)).is_some() {
1374 if !std::thread::panicking() {
1375 bg_processor.stop().unwrap();
1380 fn test_channel_manager_persist_error() {
1381 // Test that if we encounter an error during manager persistence, the thread panics.
1382 let (_, nodes) = create_nodes(2, "test_persist_error");
1383 open_channel!(nodes[0], nodes[1], 100000);
1385 let data_dir = nodes[0].kv_store.get_data_dir();
1386 let persister = Arc::new(Persister::new(data_dir).with_manager_error(std::io::ErrorKind::Other, "test"));
1387 let event_handler = |_: _| {};
1388 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()));
1389 match bg_processor.join() {
1390 Ok(_) => panic!("Expected error persisting manager"),
1392 assert_eq!(e.kind(), std::io::ErrorKind::Other);
1393 assert_eq!(e.get_ref().unwrap().to_string(), "test");
1399 #[cfg(feature = "futures")]
1400 async fn test_channel_manager_persist_error_async() {
1401 // Test that if we encounter an error during manager persistence, the thread panics.
1402 let (_, nodes) = create_nodes(2, "test_persist_error_sync");
1403 open_channel!(nodes[0], nodes[1], 100000);
1405 let data_dir = nodes[0].kv_store.get_data_dir();
1406 let persister = Arc::new(Persister::new(data_dir).with_manager_error(std::io::ErrorKind::Other, "test"));
1408 let bp_future = super::process_events_async(
1409 persister, |_: _| {async {}}, nodes[0].chain_monitor.clone(), nodes[0].node.clone(),
1410 nodes[0].rapid_gossip_sync(), nodes[0].peer_manager.clone(), nodes[0].logger.clone(),
1411 Some(nodes[0].scorer.clone()), move |dur: Duration| {
1412 Box::pin(async move {
1413 tokio::time::sleep(dur).await;
1418 match bp_future.await {
1419 Ok(_) => panic!("Expected error persisting manager"),
1421 assert_eq!(e.kind(), std::io::ErrorKind::Other);
1422 assert_eq!(e.get_ref().unwrap().to_string(), "test");
1428 fn test_network_graph_persist_error() {
1429 // Test that if we encounter an error during network graph persistence, an error gets returned.
1430 let (_, nodes) = create_nodes(2, "test_persist_network_graph_error");
1431 let data_dir = nodes[0].kv_store.get_data_dir();
1432 let persister = Arc::new(Persister::new(data_dir).with_graph_error(std::io::ErrorKind::Other, "test"));
1433 let event_handler = |_: _| {};
1434 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()));
1436 match bg_processor.stop() {
1437 Ok(_) => panic!("Expected error persisting network graph"),
1439 assert_eq!(e.kind(), std::io::ErrorKind::Other);
1440 assert_eq!(e.get_ref().unwrap().to_string(), "test");
1446 fn test_scorer_persist_error() {
1447 // Test that if we encounter an error during scorer persistence, an error gets returned.
1448 let (_, nodes) = create_nodes(2, "test_persist_scorer_error");
1449 let data_dir = nodes[0].kv_store.get_data_dir();
1450 let persister = Arc::new(Persister::new(data_dir).with_scorer_error(std::io::ErrorKind::Other, "test"));
1451 let event_handler = |_: _| {};
1452 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()));
1454 match bg_processor.stop() {
1455 Ok(_) => panic!("Expected error persisting scorer"),
1457 assert_eq!(e.kind(), std::io::ErrorKind::Other);
1458 assert_eq!(e.get_ref().unwrap().to_string(), "test");
1464 fn test_background_event_handling() {
1465 let (_, mut nodes) = create_nodes(2, "test_background_event_handling");
1466 let channel_value = 100000;
1467 let data_dir = nodes[0].kv_store.get_data_dir();
1468 let persister = Arc::new(Persister::new(data_dir.clone()));
1470 // Set up a background event handler for FundingGenerationReady events.
1471 let (funding_generation_send, funding_generation_recv) = std::sync::mpsc::sync_channel(1);
1472 let (channel_pending_send, channel_pending_recv) = std::sync::mpsc::sync_channel(1);
1473 let event_handler = move |event: Event| match event {
1474 Event::FundingGenerationReady { .. } => funding_generation_send.send(handle_funding_generation_ready!(event, channel_value)).unwrap(),
1475 Event::ChannelPending { .. } => channel_pending_send.send(()).unwrap(),
1476 Event::ChannelReady { .. } => {},
1477 _ => panic!("Unexpected event: {:?}", event),
1480 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()));
1482 // Open a channel and check that the FundingGenerationReady event was handled.
1483 begin_open_channel!(nodes[0], nodes[1], channel_value);
1484 let (temporary_channel_id, funding_tx) = funding_generation_recv
1485 .recv_timeout(Duration::from_secs(EVENT_DEADLINE))
1486 .expect("FundingGenerationReady not handled within deadline");
1487 nodes[0].node.funding_transaction_generated(&temporary_channel_id, &nodes[1].node.get_our_node_id(), funding_tx.clone()).unwrap();
1488 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()));
1489 get_event!(nodes[1], Event::ChannelPending);
1490 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()));
1491 let _ = channel_pending_recv.recv_timeout(Duration::from_secs(EVENT_DEADLINE))
1492 .expect("ChannelPending not handled within deadline");
1494 // Confirm the funding transaction.
1495 confirm_transaction(&mut nodes[0], &funding_tx);
1496 let as_funding = get_event_msg!(nodes[0], MessageSendEvent::SendChannelReady, nodes[1].node.get_our_node_id());
1497 confirm_transaction(&mut nodes[1], &funding_tx);
1498 let bs_funding = get_event_msg!(nodes[1], MessageSendEvent::SendChannelReady, nodes[0].node.get_our_node_id());
1499 nodes[0].node.handle_channel_ready(&nodes[1].node.get_our_node_id(), &bs_funding);
1500 let _as_channel_update = get_event_msg!(nodes[0], MessageSendEvent::SendChannelUpdate, nodes[1].node.get_our_node_id());
1501 nodes[1].node.handle_channel_ready(&nodes[0].node.get_our_node_id(), &as_funding);
1502 let _bs_channel_update = get_event_msg!(nodes[1], MessageSendEvent::SendChannelUpdate, nodes[0].node.get_our_node_id());
1504 if !std::thread::panicking() {
1505 bg_processor.stop().unwrap();
1508 // Set up a background event handler for SpendableOutputs events.
1509 let (sender, receiver) = std::sync::mpsc::sync_channel(1);
1510 let event_handler = move |event: Event| match event {
1511 Event::SpendableOutputs { .. } => sender.send(event).unwrap(),
1512 Event::ChannelReady { .. } => {},
1513 Event::ChannelClosed { .. } => {},
1514 _ => panic!("Unexpected event: {:?}", event),
1516 let persister = Arc::new(Persister::new(data_dir));
1517 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()));
1519 // Force close the channel and check that the SpendableOutputs event was handled.
1520 nodes[0].node.force_close_broadcasting_latest_txn(&nodes[0].node.list_channels()[0].channel_id, &nodes[1].node.get_our_node_id()).unwrap();
1521 let commitment_tx = nodes[0].tx_broadcaster.txn_broadcasted.lock().unwrap().pop().unwrap();
1522 confirm_transaction_depth(&mut nodes[0], &commitment_tx, BREAKDOWN_TIMEOUT as u32);
1524 let event = receiver
1525 .recv_timeout(Duration::from_secs(EVENT_DEADLINE))
1526 .expect("Events not handled within deadline");
1528 Event::SpendableOutputs { .. } => {},
1529 _ => panic!("Unexpected event: {:?}", event),
1532 if !std::thread::panicking() {
1533 bg_processor.stop().unwrap();
1538 fn test_scorer_persistence() {
1539 let (_, nodes) = create_nodes(2, "test_scorer_persistence");
1540 let data_dir = nodes[0].kv_store.get_data_dir();
1541 let persister = Arc::new(Persister::new(data_dir));
1542 let event_handler = |_: _| {};
1543 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()));
1546 let log_entries = nodes[0].logger.lines.lock().unwrap();
1547 let expected_log = "Persisting scorer".to_string();
1548 if log_entries.get(&("lightning_background_processor".to_string(), expected_log)).is_some() {
1553 if !std::thread::panicking() {
1554 bg_processor.stop().unwrap();
1558 macro_rules! do_test_not_pruning_network_graph_until_graph_sync_completion {
1559 ($nodes: expr, $receive: expr, $sleep: expr) => {
1560 let features = ChannelFeatures::empty();
1561 $nodes[0].network_graph.add_channel_from_partial_announcement(
1562 42, 53, features, $nodes[0].node.get_our_node_id(), $nodes[1].node.get_our_node_id()
1563 ).expect("Failed to update channel from partial announcement");
1564 let original_graph_description = $nodes[0].network_graph.to_string();
1565 assert!(original_graph_description.contains("42: features: 0000, node_one:"));
1566 assert_eq!($nodes[0].network_graph.read_only().channels().len(), 1);
1570 let log_entries = $nodes[0].logger.lines.lock().unwrap();
1571 let loop_counter = "Calling ChannelManager's timer_tick_occurred".to_string();
1572 if *log_entries.get(&("lightning_background_processor".to_string(), loop_counter))
1575 // Wait until the loop has gone around at least twice.
1580 let initialization_input = vec![
1581 76, 68, 75, 1, 111, 226, 140, 10, 182, 241, 179, 114, 193, 166, 162, 70, 174, 99, 247,
1582 79, 147, 30, 131, 101, 225, 90, 8, 156, 104, 214, 25, 0, 0, 0, 0, 0, 97, 227, 98, 218,
1583 0, 0, 0, 4, 2, 22, 7, 207, 206, 25, 164, 197, 231, 230, 231, 56, 102, 61, 250, 251,
1584 187, 172, 38, 46, 79, 247, 108, 44, 155, 48, 219, 238, 252, 53, 192, 6, 67, 2, 36, 125,
1585 157, 176, 223, 175, 234, 116, 94, 248, 201, 225, 97, 235, 50, 47, 115, 172, 63, 136,
1586 88, 216, 115, 11, 111, 217, 114, 84, 116, 124, 231, 107, 2, 158, 1, 242, 121, 152, 106,
1587 204, 131, 186, 35, 93, 70, 216, 10, 237, 224, 183, 89, 95, 65, 3, 83, 185, 58, 138,
1588 181, 64, 187, 103, 127, 68, 50, 2, 201, 19, 17, 138, 136, 149, 185, 226, 156, 137, 175,
1589 110, 32, 237, 0, 217, 90, 31, 100, 228, 149, 46, 219, 175, 168, 77, 4, 143, 38, 128,
1590 76, 97, 0, 0, 0, 2, 0, 0, 255, 8, 153, 192, 0, 2, 27, 0, 0, 0, 1, 0, 0, 255, 2, 68,
1591 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,
1592 0, 0, 0, 1, 0, 0, 0, 0, 58, 85, 116, 216, 255, 8, 153, 192, 0, 2, 27, 0, 0, 25, 0, 0,
1593 0, 1, 0, 0, 0, 125, 255, 2, 68, 226, 0, 6, 11, 0, 1, 5, 0, 0, 0, 0, 29, 129, 25, 192,
1595 $nodes[0].rapid_gossip_sync.update_network_graph_no_std(&initialization_input[..], Some(1642291930)).unwrap();
1597 // this should have added two channels and pruned the previous one.
1598 assert_eq!($nodes[0].network_graph.read_only().channels().len(), 2);
1600 $receive.expect("Network graph not pruned within deadline");
1602 // all channels should now be pruned
1603 assert_eq!($nodes[0].network_graph.read_only().channels().len(), 0);
1608 fn test_not_pruning_network_graph_until_graph_sync_completion() {
1609 let (sender, receiver) = std::sync::mpsc::sync_channel(1);
1611 let (_, nodes) = create_nodes(2, "test_not_pruning_network_graph_until_graph_sync_completion");
1612 let data_dir = nodes[0].kv_store.get_data_dir();
1613 let persister = Arc::new(Persister::new(data_dir).with_graph_persistence_notifier(sender));
1615 let event_handler = |_: _| {};
1616 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()));
1618 do_test_not_pruning_network_graph_until_graph_sync_completion!(nodes,
1619 receiver.recv_timeout(Duration::from_secs(super::FIRST_NETWORK_PRUNE_TIMER * 5)),
1620 std::thread::sleep(Duration::from_millis(1)));
1622 background_processor.stop().unwrap();
1626 #[cfg(feature = "futures")]
1627 async fn test_not_pruning_network_graph_until_graph_sync_completion_async() {
1628 let (sender, receiver) = std::sync::mpsc::sync_channel(1);
1630 let (_, nodes) = create_nodes(2, "test_not_pruning_network_graph_until_graph_sync_completion_async");
1631 let data_dir = nodes[0].kv_store.get_data_dir();
1632 let persister = Arc::new(Persister::new(data_dir).with_graph_persistence_notifier(sender));
1634 let (exit_sender, exit_receiver) = tokio::sync::watch::channel(());
1635 let bp_future = super::process_events_async(
1636 persister, |_: _| {async {}}, nodes[0].chain_monitor.clone(), nodes[0].node.clone(),
1637 nodes[0].rapid_gossip_sync(), nodes[0].peer_manager.clone(), nodes[0].logger.clone(),
1638 Some(nodes[0].scorer.clone()), move |dur: Duration| {
1639 let mut exit_receiver = exit_receiver.clone();
1640 Box::pin(async move {
1642 _ = tokio::time::sleep(dur) => false,
1643 _ = exit_receiver.changed() => true,
1649 let t1 = tokio::spawn(bp_future);
1650 let t2 = tokio::spawn(async move {
1651 do_test_not_pruning_network_graph_until_graph_sync_completion!(nodes, {
1654 tokio::time::sleep(Duration::from_secs(super::FIRST_NETWORK_PRUNE_TIMER)).await;
1655 if let Ok(()) = receiver.try_recv() { break Ok::<(), ()>(()); }
1659 }, tokio::time::sleep(Duration::from_millis(1)).await);
1660 exit_sender.send(()).unwrap();
1662 let (r1, r2) = tokio::join!(t1, t2);
1663 r1.unwrap().unwrap();
1667 macro_rules! do_test_payment_path_scoring {
1668 ($nodes: expr, $receive: expr) => {
1669 // Ensure that we update the scorer when relevant events are processed. In this case, we ensure
1670 // that we update the scorer upon a payment path succeeding (note that the channel must be
1671 // public or else we won't score it).
1672 // A background event handler for FundingGenerationReady events must be hooked up to a
1673 // running background processor.
1674 let scored_scid = 4242;
1675 let secp_ctx = Secp256k1::new();
1676 let node_1_privkey = SecretKey::from_slice(&[42; 32]).unwrap();
1677 let node_1_id = PublicKey::from_secret_key(&secp_ctx, &node_1_privkey);
1679 let path = Path { hops: vec![RouteHop {
1681 node_features: NodeFeatures::empty(),
1682 short_channel_id: scored_scid,
1683 channel_features: ChannelFeatures::empty(),
1685 cltv_expiry_delta: MIN_CLTV_EXPIRY_DELTA as u32,
1686 maybe_announced_channel: true,
1687 }], blinded_tail: None };
1689 $nodes[0].scorer.lock().unwrap().expect(TestResult::PaymentFailure { path: path.clone(), short_channel_id: scored_scid });
1690 $nodes[0].node.push_pending_event(Event::PaymentPathFailed {
1692 payment_hash: PaymentHash([42; 32]),
1693 payment_failed_permanently: false,
1694 failure: PathFailure::OnPath { network_update: None },
1696 short_channel_id: Some(scored_scid),
1698 let event = $receive.expect("PaymentPathFailed not handled within deadline");
1700 Event::PaymentPathFailed { .. } => {},
1701 _ => panic!("Unexpected event"),
1704 // Ensure we'll score payments that were explicitly failed back by the destination as
1706 $nodes[0].scorer.lock().unwrap().expect(TestResult::ProbeSuccess { path: path.clone() });
1707 $nodes[0].node.push_pending_event(Event::PaymentPathFailed {
1709 payment_hash: PaymentHash([42; 32]),
1710 payment_failed_permanently: true,
1711 failure: PathFailure::OnPath { network_update: None },
1713 short_channel_id: None,
1715 let event = $receive.expect("PaymentPathFailed not handled within deadline");
1717 Event::PaymentPathFailed { .. } => {},
1718 _ => panic!("Unexpected event"),
1721 $nodes[0].scorer.lock().unwrap().expect(TestResult::PaymentSuccess { path: path.clone() });
1722 $nodes[0].node.push_pending_event(Event::PaymentPathSuccessful {
1723 payment_id: PaymentId([42; 32]),
1727 let event = $receive.expect("PaymentPathSuccessful not handled within deadline");
1729 Event::PaymentPathSuccessful { .. } => {},
1730 _ => panic!("Unexpected event"),
1733 $nodes[0].scorer.lock().unwrap().expect(TestResult::ProbeSuccess { path: path.clone() });
1734 $nodes[0].node.push_pending_event(Event::ProbeSuccessful {
1735 payment_id: PaymentId([42; 32]),
1736 payment_hash: PaymentHash([42; 32]),
1739 let event = $receive.expect("ProbeSuccessful not handled within deadline");
1741 Event::ProbeSuccessful { .. } => {},
1742 _ => panic!("Unexpected event"),
1745 $nodes[0].scorer.lock().unwrap().expect(TestResult::ProbeFailure { path: path.clone() });
1746 $nodes[0].node.push_pending_event(Event::ProbeFailed {
1747 payment_id: PaymentId([42; 32]),
1748 payment_hash: PaymentHash([42; 32]),
1750 short_channel_id: Some(scored_scid),
1752 let event = $receive.expect("ProbeFailure not handled within deadline");
1754 Event::ProbeFailed { .. } => {},
1755 _ => panic!("Unexpected event"),
1761 fn test_payment_path_scoring() {
1762 let (sender, receiver) = std::sync::mpsc::sync_channel(1);
1763 let event_handler = move |event: Event| match event {
1764 Event::PaymentPathFailed { .. } => sender.send(event).unwrap(),
1765 Event::PaymentPathSuccessful { .. } => sender.send(event).unwrap(),
1766 Event::ProbeSuccessful { .. } => sender.send(event).unwrap(),
1767 Event::ProbeFailed { .. } => sender.send(event).unwrap(),
1768 _ => panic!("Unexpected event: {:?}", event),
1771 let (_, nodes) = create_nodes(1, "test_payment_path_scoring");
1772 let data_dir = nodes[0].kv_store.get_data_dir();
1773 let persister = Arc::new(Persister::new(data_dir));
1774 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()));
1776 do_test_payment_path_scoring!(nodes, receiver.recv_timeout(Duration::from_secs(EVENT_DEADLINE)));
1778 if !std::thread::panicking() {
1779 bg_processor.stop().unwrap();
1782 let log_entries = nodes[0].logger.lines.lock().unwrap();
1783 let expected_log = "Persisting scorer after update".to_string();
1784 assert_eq!(*log_entries.get(&("lightning_background_processor".to_string(), expected_log)).unwrap(), 5);
1788 #[cfg(feature = "futures")]
1789 async fn test_payment_path_scoring_async() {
1790 let (sender, mut receiver) = tokio::sync::mpsc::channel(1);
1791 let event_handler = move |event: Event| {
1792 let sender_ref = sender.clone();
1795 Event::PaymentPathFailed { .. } => { sender_ref.send(event).await.unwrap() },
1796 Event::PaymentPathSuccessful { .. } => { sender_ref.send(event).await.unwrap() },
1797 Event::ProbeSuccessful { .. } => { sender_ref.send(event).await.unwrap() },
1798 Event::ProbeFailed { .. } => { sender_ref.send(event).await.unwrap() },
1799 _ => panic!("Unexpected event: {:?}", event),
1804 let (_, nodes) = create_nodes(1, "test_payment_path_scoring_async");
1805 let data_dir = nodes[0].kv_store.get_data_dir();
1806 let persister = Arc::new(Persister::new(data_dir));
1808 let (exit_sender, exit_receiver) = tokio::sync::watch::channel(());
1810 let bp_future = super::process_events_async(
1811 persister, event_handler, nodes[0].chain_monitor.clone(), nodes[0].node.clone(),
1812 nodes[0].no_gossip_sync(), nodes[0].peer_manager.clone(), nodes[0].logger.clone(),
1813 Some(nodes[0].scorer.clone()), move |dur: Duration| {
1814 let mut exit_receiver = exit_receiver.clone();
1815 Box::pin(async move {
1817 _ = tokio::time::sleep(dur) => false,
1818 _ = exit_receiver.changed() => true,
1823 let t1 = tokio::spawn(bp_future);
1824 let t2 = tokio::spawn(async move {
1825 do_test_payment_path_scoring!(nodes, receiver.recv().await);
1826 exit_sender.send(()).unwrap();
1828 let log_entries = nodes[0].logger.lines.lock().unwrap();
1829 let expected_log = "Persisting scorer after update".to_string();
1830 assert_eq!(*log_entries.get(&("lightning_background_processor".to_string(), expected_log)).unwrap(), 5);
1833 let (r1, r2) = tokio::join!(t1, t2);
1834 r1.unwrap().unwrap();