1 //! Utilities that take care of tasks that (1) need to happen periodically to keep Rust-Lightning
2 //! running properly, and (2) either can or should be run in the background. See docs for
3 //! [`BackgroundProcessor`] for more details on the nitty-gritty.
5 // Prefix these with `rustdoc::` when we update our MSRV to be >= 1.52 to remove warnings.
6 #![deny(broken_intra_doc_links)]
7 #![deny(private_intra_doc_links)]
10 #![cfg_attr(not(feature = "futures"), deny(unsafe_code))]
12 #![cfg_attr(docsrs, feature(doc_auto_cfg))]
14 #![cfg_attr(all(not(feature = "std"), not(test)), no_std)]
16 #[cfg(any(test, feature = "std"))]
19 #[cfg(not(feature = "std"))]
22 #[macro_use] extern crate lightning;
23 extern crate lightning_rapid_gossip_sync;
26 use lightning::chain::chaininterface::{BroadcasterInterface, FeeEstimator};
27 use lightning::chain::chainmonitor::{ChainMonitor, Persist};
28 use lightning::sign::{EntropySource, NodeSigner, SignerProvider};
29 use lightning::events::{Event, PathFailure};
30 #[cfg(feature = "std")]
31 use lightning::events::{EventHandler, EventsProvider};
32 use lightning::ln::channelmanager::ChannelManager;
33 use lightning::ln::peer_handler::APeerManager;
34 use lightning::routing::gossip::{NetworkGraph, P2PGossipSync};
35 use lightning::routing::utxo::UtxoLookup;
36 use lightning::routing::router::Router;
37 use lightning::routing::scoring::{Score, WriteableScore};
38 use lightning::util::logger::Logger;
39 use lightning::util::persist::Persister;
40 #[cfg(feature = "std")]
41 use lightning::util::wakers::Sleeper;
42 use lightning_rapid_gossip_sync::RapidGossipSync;
45 use core::time::Duration;
47 #[cfg(feature = "std")]
49 #[cfg(feature = "std")]
50 use core::sync::atomic::{AtomicBool, Ordering};
51 #[cfg(feature = "std")]
52 use std::thread::{self, JoinHandle};
53 #[cfg(feature = "std")]
54 use std::time::Instant;
56 #[cfg(not(feature = "std"))]
59 /// `BackgroundProcessor` takes care of tasks that (1) need to happen periodically to keep
60 /// Rust-Lightning running properly, and (2) either can or should be run in the background. Its
61 /// responsibilities are:
62 /// * Processing [`Event`]s with a user-provided [`EventHandler`].
63 /// * Monitoring whether the [`ChannelManager`] needs to be re-persisted to disk, and if so,
64 /// writing it to disk/backups by invoking the callback given to it at startup.
65 /// [`ChannelManager`] persistence should be done in the background.
66 /// * Calling [`ChannelManager::timer_tick_occurred`], [`ChainMonitor::rebroadcast_pending_claims`]
67 /// and [`PeerManager::timer_tick_occurred`] at the appropriate intervals.
68 /// * Calling [`NetworkGraph::remove_stale_channels_and_tracking`] (if a [`GossipSync`] with a
69 /// [`NetworkGraph`] is provided to [`BackgroundProcessor::start`]).
71 /// It will also call [`PeerManager::process_events`] periodically though this shouldn't be relied
72 /// upon as doing so may result in high latency.
76 /// If [`ChannelManager`] persistence fails and the persisted manager becomes out-of-date, then
77 /// there is a risk of channels force-closing on startup when the manager realizes it's outdated.
78 /// However, as long as [`ChannelMonitor`] backups are sound, no funds besides those used for
79 /// unilateral chain closure fees are at risk.
81 /// [`ChannelMonitor`]: lightning::chain::channelmonitor::ChannelMonitor
82 /// [`Event`]: lightning::events::Event
83 /// [`PeerManager::timer_tick_occurred`]: lightning::ln::peer_handler::PeerManager::timer_tick_occurred
84 /// [`PeerManager::process_events`]: lightning::ln::peer_handler::PeerManager::process_events
85 #[cfg(feature = "std")]
86 #[must_use = "BackgroundProcessor will immediately stop on drop. It should be stored until shutdown."]
87 pub struct BackgroundProcessor {
88 stop_thread: Arc<AtomicBool>,
89 thread_handle: Option<JoinHandle<Result<(), std::io::Error>>>,
93 const FRESHNESS_TIMER: u64 = 60;
95 const FRESHNESS_TIMER: u64 = 1;
97 #[cfg(all(not(test), not(debug_assertions)))]
98 const PING_TIMER: u64 = 10;
99 /// Signature operations take a lot longer without compiler optimisations.
100 /// Increasing the ping timer allows for this but slower devices will be disconnected if the
101 /// timeout is reached.
102 #[cfg(all(not(test), debug_assertions))]
103 const PING_TIMER: u64 = 30;
105 const PING_TIMER: u64 = 1;
107 /// Prune the network graph of stale entries hourly.
108 const NETWORK_PRUNE_TIMER: u64 = 60 * 60;
111 const SCORER_PERSIST_TIMER: u64 = 30;
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 fn update_scorer<'a, S: 'static + Deref<Target = SC> + Send + Sync, SC: 'a + WriteableScore<'a>>(
240 scorer: &'a S, event: &Event
242 let mut score = scorer.lock();
244 Event::PaymentPathFailed { ref path, short_channel_id: Some(scid), .. } => {
245 score.payment_path_failed(path, *scid);
247 Event::PaymentPathFailed { ref path, payment_failed_permanently: true, .. } => {
248 // Reached if the destination explicitly failed it back. We treat this as a successful probe
249 // because the payment made it all the way to the destination with sufficient liquidity.
250 score.probe_successful(path);
252 Event::PaymentPathSuccessful { path, .. } => {
253 score.payment_path_successful(path);
255 Event::ProbeSuccessful { path, .. } => {
256 score.probe_successful(path);
258 Event::ProbeFailed { path, short_channel_id: Some(scid), .. } => {
259 score.probe_failed(path, *scid);
265 macro_rules! define_run_body {
266 ($persister: ident, $chain_monitor: ident, $process_chain_monitor_events: expr,
267 $channel_manager: ident, $process_channel_manager_events: expr,
268 $gossip_sync: ident, $peer_manager: ident, $logger: ident, $scorer: ident,
269 $loop_exit_check: expr, $await: expr, $get_timer: expr, $timer_elapsed: expr,
270 $check_slow_await: expr)
272 log_trace!($logger, "Calling ChannelManager's timer_tick_occurred on startup");
273 $channel_manager.timer_tick_occurred();
274 log_trace!($logger, "Rebroadcasting monitor's pending claims on startup");
275 $chain_monitor.rebroadcast_pending_claims();
277 let mut last_freshness_call = $get_timer(FRESHNESS_TIMER);
278 let mut last_ping_call = $get_timer(PING_TIMER);
279 let mut last_prune_call = $get_timer(FIRST_NETWORK_PRUNE_TIMER);
280 let mut last_scorer_persist_call = $get_timer(SCORER_PERSIST_TIMER);
281 let mut last_rebroadcast_call = $get_timer(REBROADCAST_TIMER);
282 let mut have_pruned = false;
285 $process_channel_manager_events;
286 $process_chain_monitor_events;
288 // Note that the PeerManager::process_events may block on ChannelManager's locks,
289 // hence it comes last here. When the ChannelManager finishes whatever it's doing,
290 // we want to ensure we get into `persist_manager` as quickly as we can, especially
291 // without running the normal event processing above and handing events to users.
293 // Specifically, on an *extremely* slow machine, we may see ChannelManager start
294 // processing a message effectively at any point during this loop. In order to
295 // minimize the time between such processing completing and persisting the updated
296 // ChannelManager, we want to minimize methods blocking on a ChannelManager
297 // generally, and as a fallback place such blocking only immediately before
299 $peer_manager.as_ref().process_events();
301 // Exit the loop if the background processor was requested to stop.
302 if $loop_exit_check {
303 log_trace!($logger, "Terminating background processor.");
307 // We wait up to 100ms, but track how long it takes to detect being put to sleep,
308 // see `await_start`'s use below.
309 let mut await_start = None;
310 if $check_slow_await { await_start = Some($get_timer(1)); }
311 let updates_available = $await;
312 let await_slow = if $check_slow_await { $timer_elapsed(&mut await_start.unwrap(), 1) } else { false };
314 // Exit the loop if the background processor was requested to stop.
315 if $loop_exit_check {
316 log_trace!($logger, "Terminating background processor.");
320 if updates_available {
321 log_trace!($logger, "Persisting ChannelManager...");
322 $persister.persist_manager(&*$channel_manager)?;
323 log_trace!($logger, "Done persisting ChannelManager.");
325 if $timer_elapsed(&mut last_freshness_call, FRESHNESS_TIMER) {
326 log_trace!($logger, "Calling ChannelManager's timer_tick_occurred");
327 $channel_manager.timer_tick_occurred();
328 last_freshness_call = $get_timer(FRESHNESS_TIMER);
331 // On various platforms, we may be starved of CPU cycles for several reasons.
332 // E.g. on iOS, if we've been in the background, we will be entirely paused.
333 // Similarly, if we're on a desktop platform and the device has been asleep, we
334 // may not get any cycles.
335 // We detect this by checking if our max-100ms-sleep, above, ran longer than a
336 // full second, at which point we assume sockets may have been killed (they
337 // appear to be at least on some platforms, even if it has only been a second).
338 // Note that we have to take care to not get here just because user event
339 // processing was slow at the top of the loop. For example, the sample client
340 // may call Bitcoin Core RPCs during event handling, which very often takes
341 // more than a handful of seconds to complete, and shouldn't disconnect all our
343 log_trace!($logger, "100ms sleep took more than a second, disconnecting peers.");
344 $peer_manager.as_ref().disconnect_all_peers();
345 last_ping_call = $get_timer(PING_TIMER);
346 } else if $timer_elapsed(&mut last_ping_call, PING_TIMER) {
347 log_trace!($logger, "Calling PeerManager's timer_tick_occurred");
348 $peer_manager.as_ref().timer_tick_occurred();
349 last_ping_call = $get_timer(PING_TIMER);
352 // Note that we want to run a graph prune once not long after startup before
353 // falling back to our usual hourly prunes. This avoids short-lived clients never
354 // pruning their network graph. We run once 60 seconds after startup before
355 // continuing our normal cadence.
356 let prune_timer = if have_pruned { NETWORK_PRUNE_TIMER } else { FIRST_NETWORK_PRUNE_TIMER };
357 if $timer_elapsed(&mut last_prune_call, prune_timer) {
358 // The network graph must not be pruned while rapid sync completion is pending
359 if let Some(network_graph) = $gossip_sync.prunable_network_graph() {
360 #[cfg(feature = "std")] {
361 log_trace!($logger, "Pruning and persisting network graph.");
362 network_graph.remove_stale_channels_and_tracking();
364 #[cfg(not(feature = "std"))] {
365 log_warn!($logger, "Not pruning network graph, consider enabling `std` or doing so manually with remove_stale_channels_and_tracking_with_time.");
366 log_trace!($logger, "Persisting network graph.");
369 if let Err(e) = $persister.persist_graph(network_graph) {
370 log_error!($logger, "Error: Failed to persist network graph, check your disk and permissions {}", e)
375 let prune_timer = if have_pruned { NETWORK_PRUNE_TIMER } else { FIRST_NETWORK_PRUNE_TIMER };
376 last_prune_call = $get_timer(prune_timer);
379 if $timer_elapsed(&mut last_scorer_persist_call, SCORER_PERSIST_TIMER) {
380 if let Some(ref scorer) = $scorer {
381 log_trace!($logger, "Persisting scorer");
382 if let Err(e) = $persister.persist_scorer(&scorer) {
383 log_error!($logger, "Error: Failed to persist scorer, check your disk and permissions {}", e)
386 last_scorer_persist_call = $get_timer(SCORER_PERSIST_TIMER);
389 if $timer_elapsed(&mut last_rebroadcast_call, REBROADCAST_TIMER) {
390 log_trace!($logger, "Rebroadcasting monitor's pending claims");
391 $chain_monitor.rebroadcast_pending_claims();
392 last_rebroadcast_call = $get_timer(REBROADCAST_TIMER);
396 // After we exit, ensure we persist the ChannelManager one final time - this avoids
397 // some races where users quit while channel updates were in-flight, with
398 // ChannelMonitor update(s) persisted without a corresponding ChannelManager update.
399 $persister.persist_manager(&*$channel_manager)?;
401 // Persist Scorer on exit
402 if let Some(ref scorer) = $scorer {
403 $persister.persist_scorer(&scorer)?;
406 // Persist NetworkGraph on exit
407 if let Some(network_graph) = $gossip_sync.network_graph() {
408 $persister.persist_graph(network_graph)?;
415 #[cfg(feature = "futures")]
416 pub(crate) mod futures_util {
417 use core::future::Future;
418 use core::task::{Poll, Waker, RawWaker, RawWakerVTable};
420 use core::marker::Unpin;
421 pub(crate) struct Selector<
422 A: Future<Output=()> + Unpin, B: Future<Output=()> + Unpin, C: Future<Output=bool> + Unpin
428 pub(crate) enum SelectorOutput {
433 A: Future<Output=()> + Unpin, B: Future<Output=()> + Unpin, C: Future<Output=bool> + Unpin
434 > Future for Selector<A, B, C> {
435 type Output = SelectorOutput;
436 fn poll(mut self: Pin<&mut Self>, ctx: &mut core::task::Context<'_>) -> Poll<SelectorOutput> {
437 match Pin::new(&mut self.a).poll(ctx) {
438 Poll::Ready(()) => { return Poll::Ready(SelectorOutput::A); },
441 match Pin::new(&mut self.b).poll(ctx) {
442 Poll::Ready(()) => { return Poll::Ready(SelectorOutput::B); },
445 match Pin::new(&mut self.c).poll(ctx) {
446 Poll::Ready(res) => { return Poll::Ready(SelectorOutput::C(res)); },
453 // If we want to poll a future without an async context to figure out if it has completed or
454 // not without awaiting, we need a Waker, which needs a vtable...we fill it with dummy values
455 // but sadly there's a good bit of boilerplate here.
456 fn dummy_waker_clone(_: *const ()) -> RawWaker { RawWaker::new(core::ptr::null(), &DUMMY_WAKER_VTABLE) }
457 fn dummy_waker_action(_: *const ()) { }
459 const DUMMY_WAKER_VTABLE: RawWakerVTable = RawWakerVTable::new(
460 dummy_waker_clone, dummy_waker_action, dummy_waker_action, dummy_waker_action);
461 pub(crate) fn dummy_waker() -> Waker { unsafe { Waker::from_raw(RawWaker::new(core::ptr::null(), &DUMMY_WAKER_VTABLE)) } }
463 #[cfg(feature = "futures")]
464 use futures_util::{Selector, SelectorOutput, dummy_waker};
465 #[cfg(feature = "futures")]
468 /// Processes background events in a future.
470 /// `sleeper` should return a future which completes in the given amount of time and returns a
471 /// boolean indicating whether the background processing should exit. Once `sleeper` returns a
472 /// future which outputs `true`, the loop will exit and this function's future will complete.
473 /// The `sleeper` future is free to return early after it has triggered the exit condition.
475 /// See [`BackgroundProcessor::start`] for information on which actions this handles.
477 /// Requires the `futures` feature. Note that while this method is available without the `std`
478 /// feature, doing so will skip calling [`NetworkGraph::remove_stale_channels_and_tracking`],
479 /// you should call [`NetworkGraph::remove_stale_channels_and_tracking_with_time`] regularly
480 /// manually instead.
482 /// The `mobile_interruptable_platform` flag should be set if we're currently running on a
483 /// mobile device, where we may need to check for interruption of the application regularly. If you
484 /// are unsure, you should set the flag, as the performance impact of it is minimal unless there
485 /// are hundreds or thousands of simultaneous process calls running.
487 /// For example, in order to process background events in a [Tokio](https://tokio.rs/) task, you
488 /// could setup `process_events_async` like this:
490 /// # struct MyPersister {}
491 /// # impl lightning::util::persist::KVStorePersister for MyPersister {
492 /// # fn persist<W: lightning::util::ser::Writeable>(&self, key: &str, object: &W) -> lightning::io::Result<()> { Ok(()) }
494 /// # struct MyEventHandler {}
495 /// # impl MyEventHandler {
496 /// # async fn handle_event(&self, _: lightning::events::Event) {}
498 /// # #[derive(Eq, PartialEq, Clone, Hash)]
499 /// # struct MySocketDescriptor {}
500 /// # impl lightning::ln::peer_handler::SocketDescriptor for MySocketDescriptor {
501 /// # fn send_data(&mut self, _data: &[u8], _resume_read: bool) -> usize { 0 }
502 /// # fn disconnect_socket(&mut self) {}
504 /// # use std::sync::{Arc, Mutex};
505 /// # use std::sync::atomic::{AtomicBool, Ordering};
506 /// # use lightning_background_processor::{process_events_async, GossipSync};
507 /// # type MyBroadcaster = dyn lightning::chain::chaininterface::BroadcasterInterface + Send + Sync;
508 /// # type MyFeeEstimator = dyn lightning::chain::chaininterface::FeeEstimator + Send + Sync;
509 /// # type MyNodeSigner = dyn lightning::sign::NodeSigner + Send + Sync;
510 /// # type MyUtxoLookup = dyn lightning::routing::utxo::UtxoLookup + Send + Sync;
511 /// # type MyFilter = dyn lightning::chain::Filter + Send + Sync;
512 /// # type MyLogger = dyn lightning::util::logger::Logger + Send + Sync;
513 /// # type MyChainMonitor = lightning::chain::chainmonitor::ChainMonitor<lightning::sign::InMemorySigner, Arc<MyFilter>, Arc<MyBroadcaster>, Arc<MyFeeEstimator>, Arc<MyLogger>, Arc<MyPersister>>;
514 /// # type MyPeerManager = lightning::ln::peer_handler::SimpleArcPeerManager<MySocketDescriptor, MyChainMonitor, MyBroadcaster, MyFeeEstimator, MyUtxoLookup, MyLogger>;
515 /// # type MyNetworkGraph = lightning::routing::gossip::NetworkGraph<Arc<MyLogger>>;
516 /// # type MyGossipSync = lightning::routing::gossip::P2PGossipSync<Arc<MyNetworkGraph>, Arc<MyUtxoLookup>, Arc<MyLogger>>;
517 /// # type MyChannelManager = lightning::ln::channelmanager::SimpleArcChannelManager<MyChainMonitor, MyBroadcaster, MyFeeEstimator, MyLogger>;
518 /// # type MyScorer = Mutex<lightning::routing::scoring::ProbabilisticScorer<Arc<MyNetworkGraph>, Arc<MyLogger>>>;
520 /// # async fn setup_background_processing(my_persister: Arc<MyPersister>, my_event_handler: Arc<MyEventHandler>, my_chain_monitor: Arc<MyChainMonitor>, my_channel_manager: Arc<MyChannelManager>, my_gossip_sync: Arc<MyGossipSync>, my_logger: Arc<MyLogger>, my_scorer: Arc<MyScorer>, my_peer_manager: Arc<MyPeerManager>) {
521 /// let background_persister = Arc::clone(&my_persister);
522 /// let background_event_handler = Arc::clone(&my_event_handler);
523 /// let background_chain_mon = Arc::clone(&my_chain_monitor);
524 /// let background_chan_man = Arc::clone(&my_channel_manager);
525 /// let background_gossip_sync = GossipSync::p2p(Arc::clone(&my_gossip_sync));
526 /// let background_peer_man = Arc::clone(&my_peer_manager);
527 /// let background_logger = Arc::clone(&my_logger);
528 /// let background_scorer = Arc::clone(&my_scorer);
530 /// // Setup the sleeper.
531 /// let (stop_sender, stop_receiver) = tokio::sync::watch::channel(());
533 /// let sleeper = move |d| {
534 /// let mut receiver = stop_receiver.clone();
535 /// Box::pin(async move {
537 /// _ = tokio::time::sleep(d) => false,
538 /// _ = receiver.changed() => true,
543 /// let mobile_interruptable_platform = false;
545 /// let handle = tokio::spawn(async move {
546 /// process_events_async(
547 /// background_persister,
548 /// |e| background_event_handler.handle_event(e),
549 /// background_chain_mon,
550 /// background_chan_man,
551 /// background_gossip_sync,
552 /// background_peer_man,
553 /// background_logger,
554 /// Some(background_scorer),
556 /// mobile_interruptable_platform,
559 /// .expect("Failed to process events");
562 /// // Stop the background processing.
563 /// stop_sender.send(()).unwrap();
564 /// handle.await.unwrap();
567 #[cfg(feature = "futures")]
568 pub async fn process_events_async<
570 UL: 'static + Deref + Send + Sync,
571 CF: 'static + Deref + Send + Sync,
572 CW: 'static + Deref + Send + Sync,
573 T: 'static + Deref + Send + Sync,
574 ES: 'static + Deref + Send + Sync,
575 NS: 'static + Deref + Send + Sync,
576 SP: 'static + Deref + Send + Sync,
577 F: 'static + Deref + Send + Sync,
578 R: 'static + Deref + Send + Sync,
579 G: 'static + Deref<Target = NetworkGraph<L>> + Send + Sync,
580 L: 'static + Deref + Send + Sync,
581 P: 'static + Deref + Send + Sync,
582 EventHandlerFuture: core::future::Future<Output = ()>,
583 EventHandler: Fn(Event) -> EventHandlerFuture,
584 PS: 'static + Deref + Send,
585 M: 'static + Deref<Target = ChainMonitor<<SP::Target as SignerProvider>::Signer, CF, T, F, L, P>> + Send + Sync,
586 CM: 'static + Deref<Target = ChannelManager<CW, T, ES, NS, SP, F, R, L>> + Send + Sync,
587 PGS: 'static + Deref<Target = P2PGossipSync<G, UL, L>> + Send + Sync,
588 RGS: 'static + Deref<Target = RapidGossipSync<G, L>> + Send,
589 APM: APeerManager + Send + Sync,
590 PM: 'static + Deref<Target = APM> + Send + Sync,
591 S: 'static + Deref<Target = SC> + Send + Sync,
592 SC: for<'b> WriteableScore<'b>,
593 SleepFuture: core::future::Future<Output = bool> + core::marker::Unpin,
594 Sleeper: Fn(Duration) -> SleepFuture
596 persister: PS, event_handler: EventHandler, chain_monitor: M, channel_manager: CM,
597 gossip_sync: GossipSync<PGS, RGS, G, UL, L>, peer_manager: PM, logger: L, scorer: Option<S>,
598 sleeper: Sleeper, mobile_interruptable_platform: bool,
599 ) -> Result<(), lightning::io::Error>
601 UL::Target: 'static + UtxoLookup,
602 CF::Target: 'static + chain::Filter,
603 CW::Target: 'static + chain::Watch<<SP::Target as SignerProvider>::Signer>,
604 T::Target: 'static + BroadcasterInterface,
605 ES::Target: 'static + EntropySource,
606 NS::Target: 'static + NodeSigner,
607 SP::Target: 'static + SignerProvider,
608 F::Target: 'static + FeeEstimator,
609 R::Target: 'static + Router,
610 L::Target: 'static + Logger,
611 P::Target: 'static + Persist<<SP::Target as SignerProvider>::Signer>,
612 PS::Target: 'static + Persister<'a, CW, T, ES, NS, SP, F, R, L, SC>,
614 let mut should_break = false;
615 let async_event_handler = |event| {
616 let network_graph = gossip_sync.network_graph();
617 let event_handler = &event_handler;
618 let scorer = &scorer;
620 if let Some(network_graph) = network_graph {
621 handle_network_graph_update(network_graph, &event)
623 if let Some(ref scorer) = scorer {
624 update_scorer(scorer, &event);
626 event_handler(event).await;
629 define_run_body!(persister,
630 chain_monitor, chain_monitor.process_pending_events_async(async_event_handler).await,
631 channel_manager, channel_manager.process_pending_events_async(async_event_handler).await,
632 gossip_sync, peer_manager, logger, scorer, should_break, {
634 a: channel_manager.get_persistable_update_future(),
635 b: chain_monitor.get_update_future(),
636 c: sleeper(if mobile_interruptable_platform { Duration::from_millis(100) } else { Duration::from_secs(FASTEST_TIMER) }),
639 SelectorOutput::A => true,
640 SelectorOutput::B => false,
641 SelectorOutput::C(exit) => {
646 }, |t| sleeper(Duration::from_secs(t)),
647 |fut: &mut SleepFuture, _| {
648 let mut waker = dummy_waker();
649 let mut ctx = task::Context::from_waker(&mut waker);
650 match core::pin::Pin::new(fut).poll(&mut ctx) {
651 task::Poll::Ready(exit) => { should_break = exit; true },
652 task::Poll::Pending => false,
654 }, mobile_interruptable_platform)
657 #[cfg(feature = "std")]
658 impl BackgroundProcessor {
659 /// Start a background thread that takes care of responsibilities enumerated in the [top-level
662 /// The thread runs indefinitely unless the object is dropped, [`stop`] is called, or
663 /// [`Persister::persist_manager`] returns an error. In case of an error, the error is retrieved by calling
664 /// either [`join`] or [`stop`].
666 /// # Data Persistence
668 /// [`Persister::persist_manager`] is responsible for writing out the [`ChannelManager`] to disk, and/or
669 /// uploading to one or more backup services. See [`ChannelManager::write`] for writing out a
670 /// [`ChannelManager`]. See the `lightning-persister` crate for LDK's
671 /// provided implementation.
673 /// [`Persister::persist_graph`] is responsible for writing out the [`NetworkGraph`] to disk, if
674 /// [`GossipSync`] is supplied. See [`NetworkGraph::write`] for writing out a [`NetworkGraph`].
675 /// See the `lightning-persister` crate for LDK's provided implementation.
677 /// Typically, users should either implement [`Persister::persist_manager`] to never return an
678 /// error or call [`join`] and handle any error that may arise. For the latter case,
679 /// `BackgroundProcessor` must be restarted by calling `start` again after handling the error.
683 /// `event_handler` is responsible for handling events that users should be notified of (e.g.,
684 /// payment failed). [`BackgroundProcessor`] may decorate the given [`EventHandler`] with common
685 /// functionality implemented by other handlers.
686 /// * [`P2PGossipSync`] if given will update the [`NetworkGraph`] based on payment failures.
688 /// # Rapid Gossip Sync
690 /// If rapid gossip sync is meant to run at startup, pass [`RapidGossipSync`] via `gossip_sync`
691 /// to indicate that the [`BackgroundProcessor`] should not prune the [`NetworkGraph`] instance
692 /// until the [`RapidGossipSync`] instance completes its first sync.
694 /// [top-level documentation]: BackgroundProcessor
695 /// [`join`]: Self::join
696 /// [`stop`]: Self::stop
697 /// [`ChannelManager`]: lightning::ln::channelmanager::ChannelManager
698 /// [`ChannelManager::write`]: lightning::ln::channelmanager::ChannelManager#impl-Writeable
699 /// [`Persister::persist_manager`]: lightning::util::persist::Persister::persist_manager
700 /// [`Persister::persist_graph`]: lightning::util::persist::Persister::persist_graph
701 /// [`NetworkGraph`]: lightning::routing::gossip::NetworkGraph
702 /// [`NetworkGraph::write`]: lightning::routing::gossip::NetworkGraph#impl-Writeable
705 UL: 'static + Deref + Send + Sync,
706 CF: 'static + Deref + Send + Sync,
707 CW: 'static + Deref + Send + Sync,
708 T: 'static + Deref + Send + Sync,
709 ES: 'static + Deref + Send + Sync,
710 NS: 'static + Deref + Send + Sync,
711 SP: 'static + Deref + Send + Sync,
712 F: 'static + Deref + Send + Sync,
713 R: 'static + Deref + Send + Sync,
714 G: 'static + Deref<Target = NetworkGraph<L>> + Send + Sync,
715 L: 'static + Deref + Send + Sync,
716 P: 'static + Deref + Send + Sync,
717 EH: 'static + EventHandler + Send,
718 PS: 'static + Deref + Send,
719 M: 'static + Deref<Target = ChainMonitor<<SP::Target as SignerProvider>::Signer, CF, T, F, L, P>> + Send + Sync,
720 CM: 'static + Deref<Target = ChannelManager<CW, T, ES, NS, SP, F, R, L>> + Send + Sync,
721 PGS: 'static + Deref<Target = P2PGossipSync<G, UL, L>> + Send + Sync,
722 RGS: 'static + Deref<Target = RapidGossipSync<G, L>> + Send,
723 APM: APeerManager + Send + Sync,
724 PM: 'static + Deref<Target = APM> + Send + Sync,
725 S: 'static + Deref<Target = SC> + Send + Sync,
726 SC: for <'b> WriteableScore<'b>,
728 persister: PS, event_handler: EH, chain_monitor: M, channel_manager: CM,
729 gossip_sync: GossipSync<PGS, RGS, G, UL, L>, peer_manager: PM, logger: L, scorer: Option<S>,
732 UL::Target: 'static + UtxoLookup,
733 CF::Target: 'static + chain::Filter,
734 CW::Target: 'static + chain::Watch<<SP::Target as SignerProvider>::Signer>,
735 T::Target: 'static + BroadcasterInterface,
736 ES::Target: 'static + EntropySource,
737 NS::Target: 'static + NodeSigner,
738 SP::Target: 'static + SignerProvider,
739 F::Target: 'static + FeeEstimator,
740 R::Target: 'static + Router,
741 L::Target: 'static + Logger,
742 P::Target: 'static + Persist<<SP::Target as SignerProvider>::Signer>,
743 PS::Target: 'static + Persister<'a, CW, T, ES, NS, SP, F, R, L, SC>,
745 let stop_thread = Arc::new(AtomicBool::new(false));
746 let stop_thread_clone = stop_thread.clone();
747 let handle = thread::spawn(move || -> Result<(), std::io::Error> {
748 let event_handler = |event| {
749 let network_graph = gossip_sync.network_graph();
750 if let Some(network_graph) = network_graph {
751 handle_network_graph_update(network_graph, &event)
753 if let Some(ref scorer) = scorer {
754 update_scorer(scorer, &event);
756 event_handler.handle_event(event);
758 define_run_body!(persister, chain_monitor, chain_monitor.process_pending_events(&event_handler),
759 channel_manager, channel_manager.process_pending_events(&event_handler),
760 gossip_sync, peer_manager, logger, scorer, stop_thread.load(Ordering::Acquire),
761 Sleeper::from_two_futures(
762 channel_manager.get_persistable_update_future(),
763 chain_monitor.get_update_future()
764 ).wait_timeout(Duration::from_millis(100)),
765 |_| Instant::now(), |time: &Instant, dur| time.elapsed().as_secs() > dur, false)
767 Self { stop_thread: stop_thread_clone, thread_handle: Some(handle) }
770 /// Join `BackgroundProcessor`'s thread, returning any error that occurred while persisting
771 /// [`ChannelManager`].
775 /// This function panics if the background thread has panicked such as while persisting or
778 /// [`ChannelManager`]: lightning::ln::channelmanager::ChannelManager
779 pub fn join(mut self) -> Result<(), std::io::Error> {
780 assert!(self.thread_handle.is_some());
784 /// Stop `BackgroundProcessor`'s thread, returning any error that occurred while persisting
785 /// [`ChannelManager`].
789 /// This function panics if the background thread has panicked such as while persisting or
792 /// [`ChannelManager`]: lightning::ln::channelmanager::ChannelManager
793 pub fn stop(mut self) -> Result<(), std::io::Error> {
794 assert!(self.thread_handle.is_some());
795 self.stop_and_join_thread()
798 fn stop_and_join_thread(&mut self) -> Result<(), std::io::Error> {
799 self.stop_thread.store(true, Ordering::Release);
803 fn join_thread(&mut self) -> Result<(), std::io::Error> {
804 match self.thread_handle.take() {
805 Some(handle) => handle.join().unwrap(),
811 #[cfg(feature = "std")]
812 impl Drop for BackgroundProcessor {
814 self.stop_and_join_thread().unwrap();
818 #[cfg(all(feature = "std", test))]
820 use bitcoin::blockdata::block::BlockHeader;
821 use bitcoin::blockdata::constants::genesis_block;
822 use bitcoin::blockdata::locktime::PackedLockTime;
823 use bitcoin::blockdata::transaction::{Transaction, TxOut};
824 use bitcoin::network::constants::Network;
825 use bitcoin::secp256k1::{SecretKey, PublicKey, Secp256k1};
826 use lightning::chain::{BestBlock, Confirm, chainmonitor};
827 use lightning::chain::channelmonitor::ANTI_REORG_DELAY;
828 use lightning::sign::{InMemorySigner, KeysManager};
829 use lightning::chain::transaction::OutPoint;
830 use lightning::events::{Event, PathFailure, MessageSendEventsProvider, MessageSendEvent};
831 use lightning::{get_event_msg, get_event};
832 use lightning::ln::PaymentHash;
833 use lightning::ln::channelmanager;
834 use lightning::ln::channelmanager::{BREAKDOWN_TIMEOUT, ChainParameters, MIN_CLTV_EXPIRY_DELTA, PaymentId};
835 use lightning::ln::features::{ChannelFeatures, NodeFeatures};
836 use lightning::ln::msgs::{ChannelMessageHandler, Init};
837 use lightning::ln::peer_handler::{PeerManager, MessageHandler, SocketDescriptor, IgnoringMessageHandler};
838 use lightning::routing::gossip::{NetworkGraph, NodeId, P2PGossipSync};
839 use lightning::routing::router::{DefaultRouter, Path, RouteHop};
840 use lightning::routing::scoring::{ChannelUsage, Score};
841 use lightning::util::config::UserConfig;
842 use lightning::util::ser::Writeable;
843 use lightning::util::test_utils;
844 use lightning::util::persist::KVStorePersister;
845 use lightning_persister::FilesystemPersister;
846 use std::collections::VecDeque;
848 use std::path::PathBuf;
849 use std::sync::{Arc, Mutex};
850 use std::sync::mpsc::SyncSender;
851 use std::time::Duration;
852 use bitcoin::hashes::Hash;
853 use bitcoin::TxMerkleNode;
854 use lightning_rapid_gossip_sync::RapidGossipSync;
855 use super::{BackgroundProcessor, GossipSync, FRESHNESS_TIMER};
857 const EVENT_DEADLINE: u64 = 5 * FRESHNESS_TIMER;
859 #[derive(Clone, Hash, PartialEq, Eq)]
860 struct TestDescriptor{}
861 impl SocketDescriptor for TestDescriptor {
862 fn send_data(&mut self, _data: &[u8], _resume_read: bool) -> usize {
866 fn disconnect_socket(&mut self) {}
869 type ChannelManager = channelmanager::ChannelManager<Arc<ChainMonitor>, Arc<test_utils::TestBroadcaster>, Arc<KeysManager>, Arc<KeysManager>, Arc<KeysManager>, Arc<test_utils::TestFeeEstimator>, Arc<DefaultRouter<Arc<NetworkGraph<Arc<test_utils::TestLogger>>>, Arc<test_utils::TestLogger>, Arc<Mutex<TestScorer>>, (), TestScorer>>, Arc<test_utils::TestLogger>>;
871 type ChainMonitor = chainmonitor::ChainMonitor<InMemorySigner, Arc<test_utils::TestChainSource>, Arc<test_utils::TestBroadcaster>, Arc<test_utils::TestFeeEstimator>, Arc<test_utils::TestLogger>, Arc<FilesystemPersister>>;
873 type PGS = Arc<P2PGossipSync<Arc<NetworkGraph<Arc<test_utils::TestLogger>>>, Arc<test_utils::TestChainSource>, Arc<test_utils::TestLogger>>>;
874 type RGS = Arc<RapidGossipSync<Arc<NetworkGraph<Arc<test_utils::TestLogger>>>, Arc<test_utils::TestLogger>>>;
877 node: Arc<ChannelManager>,
878 p2p_gossip_sync: PGS,
879 rapid_gossip_sync: RGS,
880 peer_manager: Arc<PeerManager<TestDescriptor, Arc<test_utils::TestChannelMessageHandler>, Arc<test_utils::TestRoutingMessageHandler>, IgnoringMessageHandler, Arc<test_utils::TestLogger>, IgnoringMessageHandler, Arc<KeysManager>>>,
881 chain_monitor: Arc<ChainMonitor>,
882 persister: Arc<FilesystemPersister>,
883 tx_broadcaster: Arc<test_utils::TestBroadcaster>,
884 network_graph: Arc<NetworkGraph<Arc<test_utils::TestLogger>>>,
885 logger: Arc<test_utils::TestLogger>,
886 best_block: BestBlock,
887 scorer: Arc<Mutex<TestScorer>>,
891 fn p2p_gossip_sync(&self) -> GossipSync<PGS, RGS, Arc<NetworkGraph<Arc<test_utils::TestLogger>>>, Arc<test_utils::TestChainSource>, Arc<test_utils::TestLogger>> {
892 GossipSync::P2P(self.p2p_gossip_sync.clone())
895 fn rapid_gossip_sync(&self) -> GossipSync<PGS, RGS, Arc<NetworkGraph<Arc<test_utils::TestLogger>>>, Arc<test_utils::TestChainSource>, Arc<test_utils::TestLogger>> {
896 GossipSync::Rapid(self.rapid_gossip_sync.clone())
899 fn no_gossip_sync(&self) -> GossipSync<PGS, RGS, Arc<NetworkGraph<Arc<test_utils::TestLogger>>>, Arc<test_utils::TestChainSource>, Arc<test_utils::TestLogger>> {
906 let data_dir = self.persister.get_data_dir();
907 match fs::remove_dir_all(data_dir.clone()) {
908 Err(e) => println!("Failed to remove test persister directory {}: {}", data_dir, e),
915 graph_error: Option<(std::io::ErrorKind, &'static str)>,
916 graph_persistence_notifier: Option<SyncSender<()>>,
917 manager_error: Option<(std::io::ErrorKind, &'static str)>,
918 scorer_error: Option<(std::io::ErrorKind, &'static str)>,
919 filesystem_persister: FilesystemPersister,
923 fn new(data_dir: String) -> Self {
924 let filesystem_persister = FilesystemPersister::new(data_dir);
925 Self { graph_error: None, graph_persistence_notifier: None, manager_error: None, scorer_error: None, filesystem_persister }
928 fn with_graph_error(self, error: std::io::ErrorKind, message: &'static str) -> Self {
929 Self { graph_error: Some((error, message)), ..self }
932 fn with_graph_persistence_notifier(self, sender: SyncSender<()>) -> Self {
933 Self { graph_persistence_notifier: Some(sender), ..self }
936 fn with_manager_error(self, error: std::io::ErrorKind, message: &'static str) -> Self {
937 Self { manager_error: Some((error, message)), ..self }
940 fn with_scorer_error(self, error: std::io::ErrorKind, message: &'static str) -> Self {
941 Self { scorer_error: Some((error, message)), ..self }
945 impl KVStorePersister for Persister {
946 fn persist<W: Writeable>(&self, key: &str, object: &W) -> std::io::Result<()> {
947 if key == "manager" {
948 if let Some((error, message)) = self.manager_error {
949 return Err(std::io::Error::new(error, message))
953 if key == "network_graph" {
954 if let Some(sender) = &self.graph_persistence_notifier {
955 match sender.send(()) {
957 Err(std::sync::mpsc::SendError(())) => println!("Persister failed to notify as receiver went away."),
961 if let Some((error, message)) = self.graph_error {
962 return Err(std::io::Error::new(error, message))
967 if let Some((error, message)) = self.scorer_error {
968 return Err(std::io::Error::new(error, message))
972 self.filesystem_persister.persist(key, object)
977 event_expectations: Option<VecDeque<TestResult>>,
982 PaymentFailure { path: Path, short_channel_id: u64 },
983 PaymentSuccess { path: Path },
984 ProbeFailure { path: Path },
985 ProbeSuccess { path: Path },
990 Self { event_expectations: None }
993 fn expect(&mut self, expectation: TestResult) {
994 self.event_expectations.get_or_insert_with(VecDeque::new).push_back(expectation);
998 impl lightning::util::ser::Writeable for TestScorer {
999 fn write<W: lightning::util::ser::Writer>(&self, _: &mut W) -> Result<(), lightning::io::Error> { Ok(()) }
1002 impl Score for TestScorer {
1003 type ScoreParams = ();
1004 fn channel_penalty_msat(
1005 &self, _short_channel_id: u64, _source: &NodeId, _target: &NodeId, _usage: ChannelUsage, _score_params: &Self::ScoreParams
1006 ) -> u64 { unimplemented!(); }
1008 fn payment_path_failed(&mut self, actual_path: &Path, actual_short_channel_id: u64) {
1009 if let Some(expectations) = &mut self.event_expectations {
1010 match expectations.pop_front().unwrap() {
1011 TestResult::PaymentFailure { path, short_channel_id } => {
1012 assert_eq!(actual_path, &path);
1013 assert_eq!(actual_short_channel_id, short_channel_id);
1015 TestResult::PaymentSuccess { path } => {
1016 panic!("Unexpected successful payment path: {:?}", path)
1018 TestResult::ProbeFailure { path } => {
1019 panic!("Unexpected probe failure: {:?}", path)
1021 TestResult::ProbeSuccess { path } => {
1022 panic!("Unexpected probe success: {:?}", path)
1028 fn payment_path_successful(&mut self, actual_path: &Path) {
1029 if let Some(expectations) = &mut self.event_expectations {
1030 match expectations.pop_front().unwrap() {
1031 TestResult::PaymentFailure { path, .. } => {
1032 panic!("Unexpected payment path failure: {:?}", path)
1034 TestResult::PaymentSuccess { path } => {
1035 assert_eq!(actual_path, &path);
1037 TestResult::ProbeFailure { path } => {
1038 panic!("Unexpected probe failure: {:?}", path)
1040 TestResult::ProbeSuccess { path } => {
1041 panic!("Unexpected probe success: {:?}", path)
1047 fn probe_failed(&mut self, actual_path: &Path, _: u64) {
1048 if let Some(expectations) = &mut self.event_expectations {
1049 match expectations.pop_front().unwrap() {
1050 TestResult::PaymentFailure { path, .. } => {
1051 panic!("Unexpected payment path failure: {:?}", path)
1053 TestResult::PaymentSuccess { path } => {
1054 panic!("Unexpected payment path success: {:?}", path)
1056 TestResult::ProbeFailure { path } => {
1057 assert_eq!(actual_path, &path);
1059 TestResult::ProbeSuccess { path } => {
1060 panic!("Unexpected probe success: {:?}", path)
1065 fn probe_successful(&mut self, actual_path: &Path) {
1066 if let Some(expectations) = &mut self.event_expectations {
1067 match expectations.pop_front().unwrap() {
1068 TestResult::PaymentFailure { path, .. } => {
1069 panic!("Unexpected payment path failure: {:?}", path)
1071 TestResult::PaymentSuccess { path } => {
1072 panic!("Unexpected payment path success: {:?}", path)
1074 TestResult::ProbeFailure { path } => {
1075 panic!("Unexpected probe failure: {:?}", path)
1077 TestResult::ProbeSuccess { path } => {
1078 assert_eq!(actual_path, &path);
1085 impl Drop for TestScorer {
1086 fn drop(&mut self) {
1087 if std::thread::panicking() {
1091 if let Some(event_expectations) = &self.event_expectations {
1092 if !event_expectations.is_empty() {
1093 panic!("Unsatisfied event expectations: {:?}", event_expectations);
1099 fn get_full_filepath(filepath: String, filename: String) -> String {
1100 let mut path = PathBuf::from(filepath);
1101 path.push(filename);
1102 path.to_str().unwrap().to_string()
1105 fn create_nodes(num_nodes: usize, persist_dir: &str) -> (String, Vec<Node>) {
1106 let persist_temp_path = env::temp_dir().join(persist_dir);
1107 let persist_dir = persist_temp_path.to_string_lossy().to_string();
1108 let network = Network::Testnet;
1109 let mut nodes = Vec::new();
1110 for i in 0..num_nodes {
1111 let tx_broadcaster = Arc::new(test_utils::TestBroadcaster::new(network));
1112 let fee_estimator = Arc::new(test_utils::TestFeeEstimator { sat_per_kw: Mutex::new(253) });
1113 let logger = Arc::new(test_utils::TestLogger::with_id(format!("node {}", i)));
1114 let genesis_block = genesis_block(network);
1115 let network_graph = Arc::new(NetworkGraph::new(network, logger.clone()));
1116 let scorer = Arc::new(Mutex::new(TestScorer::new()));
1117 let seed = [i as u8; 32];
1118 let router = Arc::new(DefaultRouter::new(network_graph.clone(), logger.clone(), seed, scorer.clone(), ()));
1119 let chain_source = Arc::new(test_utils::TestChainSource::new(Network::Testnet));
1120 let persister = Arc::new(FilesystemPersister::new(format!("{}_persister_{}", &persist_dir, i)));
1121 let now = Duration::from_secs(genesis_block.header.time as u64);
1122 let keys_manager = Arc::new(KeysManager::new(&seed, now.as_secs(), now.subsec_nanos()));
1123 let chain_monitor = Arc::new(chainmonitor::ChainMonitor::new(Some(chain_source.clone()), tx_broadcaster.clone(), logger.clone(), fee_estimator.clone(), persister.clone()));
1124 let best_block = BestBlock::from_network(network);
1125 let params = ChainParameters { network, best_block };
1126 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));
1127 let p2p_gossip_sync = Arc::new(P2PGossipSync::new(network_graph.clone(), Some(chain_source.clone()), logger.clone()));
1128 let rapid_gossip_sync = Arc::new(RapidGossipSync::new(network_graph.clone(), logger.clone()));
1129 let msg_handler = MessageHandler {
1130 chan_handler: Arc::new(test_utils::TestChannelMessageHandler::new()),
1131 route_handler: Arc::new(test_utils::TestRoutingMessageHandler::new()),
1132 onion_message_handler: IgnoringMessageHandler{}, custom_message_handler: IgnoringMessageHandler{}
1134 let peer_manager = Arc::new(PeerManager::new(msg_handler, 0, &seed, logger.clone(), keys_manager.clone()));
1135 let node = Node { node: manager, p2p_gossip_sync, rapid_gossip_sync, peer_manager, chain_monitor, persister, tx_broadcaster, network_graph, logger, best_block, scorer };
1139 for i in 0..num_nodes {
1140 for j in (i+1)..num_nodes {
1141 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();
1142 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();
1146 (persist_dir, nodes)
1149 macro_rules! open_channel {
1150 ($node_a: expr, $node_b: expr, $channel_value: expr) => {{
1151 begin_open_channel!($node_a, $node_b, $channel_value);
1152 let events = $node_a.node.get_and_clear_pending_events();
1153 assert_eq!(events.len(), 1);
1154 let (temporary_channel_id, tx) = handle_funding_generation_ready!(events[0], $channel_value);
1155 $node_a.node.funding_transaction_generated(&temporary_channel_id, &$node_b.node.get_our_node_id(), tx.clone()).unwrap();
1156 $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()));
1157 get_event!($node_b, Event::ChannelPending);
1158 $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()));
1159 get_event!($node_a, Event::ChannelPending);
1164 macro_rules! begin_open_channel {
1165 ($node_a: expr, $node_b: expr, $channel_value: expr) => {{
1166 $node_a.node.create_channel($node_b.node.get_our_node_id(), $channel_value, 100, 42, None).unwrap();
1167 $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()));
1168 $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()));
1172 macro_rules! handle_funding_generation_ready {
1173 ($event: expr, $channel_value: expr) => {{
1175 Event::FundingGenerationReady { temporary_channel_id, channel_value_satoshis, ref output_script, user_channel_id, .. } => {
1176 assert_eq!(channel_value_satoshis, $channel_value);
1177 assert_eq!(user_channel_id, 42);
1179 let tx = Transaction { version: 1 as i32, lock_time: PackedLockTime(0), input: Vec::new(), output: vec![TxOut {
1180 value: channel_value_satoshis, script_pubkey: output_script.clone(),
1182 (temporary_channel_id, tx)
1184 _ => panic!("Unexpected event"),
1189 fn confirm_transaction_depth(node: &mut Node, tx: &Transaction, depth: u32) {
1190 for i in 1..=depth {
1191 let prev_blockhash = node.best_block.block_hash();
1192 let height = node.best_block.height() + 1;
1193 let header = BlockHeader { version: 0x20000000, prev_blockhash, merkle_root: TxMerkleNode::all_zeros(), time: height, bits: 42, nonce: 42 };
1194 let txdata = vec![(0, tx)];
1195 node.best_block = BestBlock::new(header.block_hash(), height);
1198 node.node.transactions_confirmed(&header, &txdata, height);
1199 node.chain_monitor.transactions_confirmed(&header, &txdata, height);
1201 x if x == depth => {
1202 node.node.best_block_updated(&header, height);
1203 node.chain_monitor.best_block_updated(&header, height);
1209 fn confirm_transaction(node: &mut Node, tx: &Transaction) {
1210 confirm_transaction_depth(node, tx, ANTI_REORG_DELAY);
1214 fn test_background_processor() {
1215 // Test that when a new channel is created, the ChannelManager needs to be re-persisted with
1216 // updates. Also test that when new updates are available, the manager signals that it needs
1217 // re-persistence and is successfully re-persisted.
1218 let (persist_dir, nodes) = create_nodes(2, "test_background_processor");
1220 // Go through the channel creation process so that each node has something to persist. Since
1221 // open_channel consumes events, it must complete before starting BackgroundProcessor to
1222 // avoid a race with processing events.
1223 let tx = open_channel!(nodes[0], nodes[1], 100000);
1225 // Initiate the background processors to watch each node.
1226 let data_dir = nodes[0].persister.get_data_dir();
1227 let persister = Arc::new(Persister::new(data_dir));
1228 let event_handler = |_: _| {};
1229 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()));
1231 macro_rules! check_persisted_data {
1232 ($node: expr, $filepath: expr) => {
1233 let mut expected_bytes = Vec::new();
1235 expected_bytes.clear();
1236 match $node.write(&mut expected_bytes) {
1238 match std::fs::read($filepath) {
1240 if bytes == expected_bytes {
1249 Err(e) => panic!("Unexpected error: {}", e)
1255 // Check that the initial channel manager data is persisted as expected.
1256 let filepath = get_full_filepath(format!("{}_persister_0", &persist_dir), "manager".to_string());
1257 check_persisted_data!(nodes[0].node, filepath.clone());
1260 if !nodes[0].node.get_persistence_condvar_value() { break }
1263 // Force-close the channel.
1264 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();
1266 // Check that the force-close updates are persisted.
1267 check_persisted_data!(nodes[0].node, filepath.clone());
1269 if !nodes[0].node.get_persistence_condvar_value() { break }
1272 // Check network graph is persisted
1273 let filepath = get_full_filepath(format!("{}_persister_0", &persist_dir), "network_graph".to_string());
1274 check_persisted_data!(nodes[0].network_graph, filepath.clone());
1276 // Check scorer is persisted
1277 let filepath = get_full_filepath(format!("{}_persister_0", &persist_dir), "scorer".to_string());
1278 check_persisted_data!(nodes[0].scorer, filepath.clone());
1280 if !std::thread::panicking() {
1281 bg_processor.stop().unwrap();
1286 fn test_timer_tick_called() {
1287 // Test that `ChannelManager::timer_tick_occurred` is called every `FRESHNESS_TIMER`,
1288 // `ChainMonitor::rebroadcast_pending_claims` is called every `REBROADCAST_TIMER`, and
1289 // `PeerManager::timer_tick_occurred` every `PING_TIMER`.
1290 let (_, nodes) = create_nodes(1, "test_timer_tick_called");
1291 let data_dir = nodes[0].persister.get_data_dir();
1292 let persister = Arc::new(Persister::new(data_dir));
1293 let event_handler = |_: _| {};
1294 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()));
1296 let log_entries = nodes[0].logger.lines.lock().unwrap();
1297 let desired_log_1 = "Calling ChannelManager's timer_tick_occurred".to_string();
1298 let desired_log_2 = "Calling PeerManager's timer_tick_occurred".to_string();
1299 let desired_log_3 = "Rebroadcasting monitor's pending claims".to_string();
1300 if log_entries.get(&("lightning_background_processor".to_string(), desired_log_1)).is_some() &&
1301 log_entries.get(&("lightning_background_processor".to_string(), desired_log_2)).is_some() &&
1302 log_entries.get(&("lightning_background_processor".to_string(), desired_log_3)).is_some() {
1307 if !std::thread::panicking() {
1308 bg_processor.stop().unwrap();
1313 fn test_channel_manager_persist_error() {
1314 // Test that if we encounter an error during manager persistence, the thread panics.
1315 let (_, nodes) = create_nodes(2, "test_persist_error");
1316 open_channel!(nodes[0], nodes[1], 100000);
1318 let data_dir = nodes[0].persister.get_data_dir();
1319 let persister = Arc::new(Persister::new(data_dir).with_manager_error(std::io::ErrorKind::Other, "test"));
1320 let event_handler = |_: _| {};
1321 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()));
1322 match bg_processor.join() {
1323 Ok(_) => panic!("Expected error persisting manager"),
1325 assert_eq!(e.kind(), std::io::ErrorKind::Other);
1326 assert_eq!(e.get_ref().unwrap().to_string(), "test");
1332 #[cfg(feature = "futures")]
1333 async fn test_channel_manager_persist_error_async() {
1334 // Test that if we encounter an error during manager persistence, the thread panics.
1335 let (_, nodes) = create_nodes(2, "test_persist_error_sync");
1336 open_channel!(nodes[0], nodes[1], 100000);
1338 let data_dir = nodes[0].persister.get_data_dir();
1339 let persister = Arc::new(Persister::new(data_dir).with_manager_error(std::io::ErrorKind::Other, "test"));
1341 let bp_future = super::process_events_async(
1342 persister, |_: _| {async {}}, nodes[0].chain_monitor.clone(), nodes[0].node.clone(),
1343 nodes[0].rapid_gossip_sync(), nodes[0].peer_manager.clone(), nodes[0].logger.clone(),
1344 Some(nodes[0].scorer.clone()), move |dur: Duration| {
1345 Box::pin(async move {
1346 tokio::time::sleep(dur).await;
1351 match bp_future.await {
1352 Ok(_) => panic!("Expected error persisting manager"),
1354 assert_eq!(e.kind(), std::io::ErrorKind::Other);
1355 assert_eq!(e.get_ref().unwrap().to_string(), "test");
1361 fn test_network_graph_persist_error() {
1362 // Test that if we encounter an error during network graph persistence, an error gets returned.
1363 let (_, nodes) = create_nodes(2, "test_persist_network_graph_error");
1364 let data_dir = nodes[0].persister.get_data_dir();
1365 let persister = Arc::new(Persister::new(data_dir).with_graph_error(std::io::ErrorKind::Other, "test"));
1366 let event_handler = |_: _| {};
1367 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()));
1369 match bg_processor.stop() {
1370 Ok(_) => panic!("Expected error persisting network graph"),
1372 assert_eq!(e.kind(), std::io::ErrorKind::Other);
1373 assert_eq!(e.get_ref().unwrap().to_string(), "test");
1379 fn test_scorer_persist_error() {
1380 // Test that if we encounter an error during scorer persistence, an error gets returned.
1381 let (_, nodes) = create_nodes(2, "test_persist_scorer_error");
1382 let data_dir = nodes[0].persister.get_data_dir();
1383 let persister = Arc::new(Persister::new(data_dir).with_scorer_error(std::io::ErrorKind::Other, "test"));
1384 let event_handler = |_: _| {};
1385 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()));
1387 match bg_processor.stop() {
1388 Ok(_) => panic!("Expected error persisting scorer"),
1390 assert_eq!(e.kind(), std::io::ErrorKind::Other);
1391 assert_eq!(e.get_ref().unwrap().to_string(), "test");
1397 fn test_background_event_handling() {
1398 let (_, mut nodes) = create_nodes(2, "test_background_event_handling");
1399 let channel_value = 100000;
1400 let data_dir = nodes[0].persister.get_data_dir();
1401 let persister = Arc::new(Persister::new(data_dir.clone()));
1403 // Set up a background event handler for FundingGenerationReady events.
1404 let (funding_generation_send, funding_generation_recv) = std::sync::mpsc::sync_channel(1);
1405 let (channel_pending_send, channel_pending_recv) = std::sync::mpsc::sync_channel(1);
1406 let event_handler = move |event: Event| match event {
1407 Event::FundingGenerationReady { .. } => funding_generation_send.send(handle_funding_generation_ready!(event, channel_value)).unwrap(),
1408 Event::ChannelPending { .. } => channel_pending_send.send(()).unwrap(),
1409 Event::ChannelReady { .. } => {},
1410 _ => panic!("Unexpected event: {:?}", event),
1413 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()));
1415 // Open a channel and check that the FundingGenerationReady event was handled.
1416 begin_open_channel!(nodes[0], nodes[1], channel_value);
1417 let (temporary_channel_id, funding_tx) = funding_generation_recv
1418 .recv_timeout(Duration::from_secs(EVENT_DEADLINE))
1419 .expect("FundingGenerationReady not handled within deadline");
1420 nodes[0].node.funding_transaction_generated(&temporary_channel_id, &nodes[1].node.get_our_node_id(), funding_tx.clone()).unwrap();
1421 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()));
1422 get_event!(nodes[1], Event::ChannelPending);
1423 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()));
1424 let _ = channel_pending_recv.recv_timeout(Duration::from_secs(EVENT_DEADLINE))
1425 .expect("ChannelPending not handled within deadline");
1427 // Confirm the funding transaction.
1428 confirm_transaction(&mut nodes[0], &funding_tx);
1429 let as_funding = get_event_msg!(nodes[0], MessageSendEvent::SendChannelReady, nodes[1].node.get_our_node_id());
1430 confirm_transaction(&mut nodes[1], &funding_tx);
1431 let bs_funding = get_event_msg!(nodes[1], MessageSendEvent::SendChannelReady, nodes[0].node.get_our_node_id());
1432 nodes[0].node.handle_channel_ready(&nodes[1].node.get_our_node_id(), &bs_funding);
1433 let _as_channel_update = get_event_msg!(nodes[0], MessageSendEvent::SendChannelUpdate, nodes[1].node.get_our_node_id());
1434 nodes[1].node.handle_channel_ready(&nodes[0].node.get_our_node_id(), &as_funding);
1435 let _bs_channel_update = get_event_msg!(nodes[1], MessageSendEvent::SendChannelUpdate, nodes[0].node.get_our_node_id());
1437 if !std::thread::panicking() {
1438 bg_processor.stop().unwrap();
1441 // Set up a background event handler for SpendableOutputs events.
1442 let (sender, receiver) = std::sync::mpsc::sync_channel(1);
1443 let event_handler = move |event: Event| match event {
1444 Event::SpendableOutputs { .. } => sender.send(event).unwrap(),
1445 Event::ChannelReady { .. } => {},
1446 Event::ChannelClosed { .. } => {},
1447 _ => panic!("Unexpected event: {:?}", event),
1449 let persister = Arc::new(Persister::new(data_dir));
1450 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()));
1452 // Force close the channel and check that the SpendableOutputs event was handled.
1453 nodes[0].node.force_close_broadcasting_latest_txn(&nodes[0].node.list_channels()[0].channel_id, &nodes[1].node.get_our_node_id()).unwrap();
1454 let commitment_tx = nodes[0].tx_broadcaster.txn_broadcasted.lock().unwrap().pop().unwrap();
1455 confirm_transaction_depth(&mut nodes[0], &commitment_tx, BREAKDOWN_TIMEOUT as u32);
1457 let event = receiver
1458 .recv_timeout(Duration::from_secs(EVENT_DEADLINE))
1459 .expect("Events not handled within deadline");
1461 Event::SpendableOutputs { .. } => {},
1462 _ => panic!("Unexpected event: {:?}", event),
1465 if !std::thread::panicking() {
1466 bg_processor.stop().unwrap();
1471 fn test_scorer_persistence() {
1472 let (_, nodes) = create_nodes(2, "test_scorer_persistence");
1473 let data_dir = nodes[0].persister.get_data_dir();
1474 let persister = Arc::new(Persister::new(data_dir));
1475 let event_handler = |_: _| {};
1476 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()));
1479 let log_entries = nodes[0].logger.lines.lock().unwrap();
1480 let expected_log = "Persisting scorer".to_string();
1481 if log_entries.get(&("lightning_background_processor".to_string(), expected_log)).is_some() {
1486 if !std::thread::panicking() {
1487 bg_processor.stop().unwrap();
1491 macro_rules! do_test_not_pruning_network_graph_until_graph_sync_completion {
1492 ($nodes: expr, $receive: expr, $sleep: expr) => {
1493 let features = ChannelFeatures::empty();
1494 $nodes[0].network_graph.add_channel_from_partial_announcement(
1495 42, 53, features, $nodes[0].node.get_our_node_id(), $nodes[1].node.get_our_node_id()
1496 ).expect("Failed to update channel from partial announcement");
1497 let original_graph_description = $nodes[0].network_graph.to_string();
1498 assert!(original_graph_description.contains("42: features: 0000, node_one:"));
1499 assert_eq!($nodes[0].network_graph.read_only().channels().len(), 1);
1503 let log_entries = $nodes[0].logger.lines.lock().unwrap();
1504 let loop_counter = "Calling ChannelManager's timer_tick_occurred".to_string();
1505 if *log_entries.get(&("lightning_background_processor".to_string(), loop_counter))
1508 // Wait until the loop has gone around at least twice.
1513 let initialization_input = vec![
1514 76, 68, 75, 1, 111, 226, 140, 10, 182, 241, 179, 114, 193, 166, 162, 70, 174, 99, 247,
1515 79, 147, 30, 131, 101, 225, 90, 8, 156, 104, 214, 25, 0, 0, 0, 0, 0, 97, 227, 98, 218,
1516 0, 0, 0, 4, 2, 22, 7, 207, 206, 25, 164, 197, 231, 230, 231, 56, 102, 61, 250, 251,
1517 187, 172, 38, 46, 79, 247, 108, 44, 155, 48, 219, 238, 252, 53, 192, 6, 67, 2, 36, 125,
1518 157, 176, 223, 175, 234, 116, 94, 248, 201, 225, 97, 235, 50, 47, 115, 172, 63, 136,
1519 88, 216, 115, 11, 111, 217, 114, 84, 116, 124, 231, 107, 2, 158, 1, 242, 121, 152, 106,
1520 204, 131, 186, 35, 93, 70, 216, 10, 237, 224, 183, 89, 95, 65, 3, 83, 185, 58, 138,
1521 181, 64, 187, 103, 127, 68, 50, 2, 201, 19, 17, 138, 136, 149, 185, 226, 156, 137, 175,
1522 110, 32, 237, 0, 217, 90, 31, 100, 228, 149, 46, 219, 175, 168, 77, 4, 143, 38, 128,
1523 76, 97, 0, 0, 0, 2, 0, 0, 255, 8, 153, 192, 0, 2, 27, 0, 0, 0, 1, 0, 0, 255, 2, 68,
1524 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,
1525 0, 0, 0, 1, 0, 0, 0, 0, 58, 85, 116, 216, 255, 8, 153, 192, 0, 2, 27, 0, 0, 25, 0, 0,
1526 0, 1, 0, 0, 0, 125, 255, 2, 68, 226, 0, 6, 11, 0, 1, 5, 0, 0, 0, 0, 29, 129, 25, 192,
1528 $nodes[0].rapid_gossip_sync.update_network_graph_no_std(&initialization_input[..], Some(1642291930)).unwrap();
1530 // this should have added two channels and pruned the previous one.
1531 assert_eq!($nodes[0].network_graph.read_only().channels().len(), 2);
1533 $receive.expect("Network graph not pruned within deadline");
1535 // all channels should now be pruned
1536 assert_eq!($nodes[0].network_graph.read_only().channels().len(), 0);
1541 fn test_not_pruning_network_graph_until_graph_sync_completion() {
1542 let (sender, receiver) = std::sync::mpsc::sync_channel(1);
1544 let (_, nodes) = create_nodes(2, "test_not_pruning_network_graph_until_graph_sync_completion");
1545 let data_dir = nodes[0].persister.get_data_dir();
1546 let persister = Arc::new(Persister::new(data_dir).with_graph_persistence_notifier(sender));
1548 let event_handler = |_: _| {};
1549 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()));
1551 do_test_not_pruning_network_graph_until_graph_sync_completion!(nodes,
1552 receiver.recv_timeout(Duration::from_secs(super::FIRST_NETWORK_PRUNE_TIMER * 5)),
1553 std::thread::sleep(Duration::from_millis(1)));
1555 background_processor.stop().unwrap();
1559 #[cfg(feature = "futures")]
1560 async fn test_not_pruning_network_graph_until_graph_sync_completion_async() {
1561 let (sender, receiver) = std::sync::mpsc::sync_channel(1);
1563 let (_, nodes) = create_nodes(2, "test_not_pruning_network_graph_until_graph_sync_completion_async");
1564 let data_dir = nodes[0].persister.get_data_dir();
1565 let persister = Arc::new(Persister::new(data_dir).with_graph_persistence_notifier(sender));
1567 let (exit_sender, exit_receiver) = tokio::sync::watch::channel(());
1568 let bp_future = super::process_events_async(
1569 persister, |_: _| {async {}}, nodes[0].chain_monitor.clone(), nodes[0].node.clone(),
1570 nodes[0].rapid_gossip_sync(), nodes[0].peer_manager.clone(), nodes[0].logger.clone(),
1571 Some(nodes[0].scorer.clone()), move |dur: Duration| {
1572 let mut exit_receiver = exit_receiver.clone();
1573 Box::pin(async move {
1575 _ = tokio::time::sleep(dur) => false,
1576 _ = exit_receiver.changed() => true,
1582 let t1 = tokio::spawn(bp_future);
1583 let t2 = tokio::spawn(async move {
1584 do_test_not_pruning_network_graph_until_graph_sync_completion!(nodes, {
1587 tokio::time::sleep(Duration::from_secs(super::FIRST_NETWORK_PRUNE_TIMER)).await;
1588 if let Ok(()) = receiver.try_recv() { break Ok::<(), ()>(()); }
1592 }, tokio::time::sleep(Duration::from_millis(1)).await);
1593 exit_sender.send(()).unwrap();
1595 let (r1, r2) = tokio::join!(t1, t2);
1596 r1.unwrap().unwrap();
1600 macro_rules! do_test_payment_path_scoring {
1601 ($nodes: expr, $receive: expr) => {
1602 // Ensure that we update the scorer when relevant events are processed. In this case, we ensure
1603 // that we update the scorer upon a payment path succeeding (note that the channel must be
1604 // public or else we won't score it).
1605 // A background event handler for FundingGenerationReady events must be hooked up to a
1606 // running background processor.
1607 let scored_scid = 4242;
1608 let secp_ctx = Secp256k1::new();
1609 let node_1_privkey = SecretKey::from_slice(&[42; 32]).unwrap();
1610 let node_1_id = PublicKey::from_secret_key(&secp_ctx, &node_1_privkey);
1612 let path = Path { hops: vec![RouteHop {
1614 node_features: NodeFeatures::empty(),
1615 short_channel_id: scored_scid,
1616 channel_features: ChannelFeatures::empty(),
1618 cltv_expiry_delta: MIN_CLTV_EXPIRY_DELTA as u32,
1619 }], blinded_tail: None };
1621 $nodes[0].scorer.lock().unwrap().expect(TestResult::PaymentFailure { path: path.clone(), short_channel_id: scored_scid });
1622 $nodes[0].node.push_pending_event(Event::PaymentPathFailed {
1624 payment_hash: PaymentHash([42; 32]),
1625 payment_failed_permanently: false,
1626 failure: PathFailure::OnPath { network_update: None },
1628 short_channel_id: Some(scored_scid),
1630 let event = $receive.expect("PaymentPathFailed not handled within deadline");
1632 Event::PaymentPathFailed { .. } => {},
1633 _ => panic!("Unexpected event"),
1636 // Ensure we'll score payments that were explicitly failed back by the destination as
1638 $nodes[0].scorer.lock().unwrap().expect(TestResult::ProbeSuccess { path: path.clone() });
1639 $nodes[0].node.push_pending_event(Event::PaymentPathFailed {
1641 payment_hash: PaymentHash([42; 32]),
1642 payment_failed_permanently: true,
1643 failure: PathFailure::OnPath { network_update: None },
1645 short_channel_id: None,
1647 let event = $receive.expect("PaymentPathFailed not handled within deadline");
1649 Event::PaymentPathFailed { .. } => {},
1650 _ => panic!("Unexpected event"),
1653 $nodes[0].scorer.lock().unwrap().expect(TestResult::PaymentSuccess { path: path.clone() });
1654 $nodes[0].node.push_pending_event(Event::PaymentPathSuccessful {
1655 payment_id: PaymentId([42; 32]),
1659 let event = $receive.expect("PaymentPathSuccessful not handled within deadline");
1661 Event::PaymentPathSuccessful { .. } => {},
1662 _ => panic!("Unexpected event"),
1665 $nodes[0].scorer.lock().unwrap().expect(TestResult::ProbeSuccess { path: path.clone() });
1666 $nodes[0].node.push_pending_event(Event::ProbeSuccessful {
1667 payment_id: PaymentId([42; 32]),
1668 payment_hash: PaymentHash([42; 32]),
1671 let event = $receive.expect("ProbeSuccessful not handled within deadline");
1673 Event::ProbeSuccessful { .. } => {},
1674 _ => panic!("Unexpected event"),
1677 $nodes[0].scorer.lock().unwrap().expect(TestResult::ProbeFailure { path: path.clone() });
1678 $nodes[0].node.push_pending_event(Event::ProbeFailed {
1679 payment_id: PaymentId([42; 32]),
1680 payment_hash: PaymentHash([42; 32]),
1682 short_channel_id: Some(scored_scid),
1684 let event = $receive.expect("ProbeFailure not handled within deadline");
1686 Event::ProbeFailed { .. } => {},
1687 _ => panic!("Unexpected event"),
1693 fn test_payment_path_scoring() {
1694 let (sender, receiver) = std::sync::mpsc::sync_channel(1);
1695 let event_handler = move |event: Event| match event {
1696 Event::PaymentPathFailed { .. } => sender.send(event).unwrap(),
1697 Event::PaymentPathSuccessful { .. } => sender.send(event).unwrap(),
1698 Event::ProbeSuccessful { .. } => sender.send(event).unwrap(),
1699 Event::ProbeFailed { .. } => sender.send(event).unwrap(),
1700 _ => panic!("Unexpected event: {:?}", event),
1703 let (_, nodes) = create_nodes(1, "test_payment_path_scoring");
1704 let data_dir = nodes[0].persister.get_data_dir();
1705 let persister = Arc::new(Persister::new(data_dir));
1706 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()));
1708 do_test_payment_path_scoring!(nodes, receiver.recv_timeout(Duration::from_secs(EVENT_DEADLINE)));
1710 if !std::thread::panicking() {
1711 bg_processor.stop().unwrap();
1716 #[cfg(feature = "futures")]
1717 async fn test_payment_path_scoring_async() {
1718 let (sender, mut receiver) = tokio::sync::mpsc::channel(1);
1719 let event_handler = move |event: Event| {
1720 let sender_ref = sender.clone();
1723 Event::PaymentPathFailed { .. } => { sender_ref.send(event).await.unwrap() },
1724 Event::PaymentPathSuccessful { .. } => { sender_ref.send(event).await.unwrap() },
1725 Event::ProbeSuccessful { .. } => { sender_ref.send(event).await.unwrap() },
1726 Event::ProbeFailed { .. } => { sender_ref.send(event).await.unwrap() },
1727 _ => panic!("Unexpected event: {:?}", event),
1732 let (_, nodes) = create_nodes(1, "test_payment_path_scoring_async");
1733 let data_dir = nodes[0].persister.get_data_dir();
1734 let persister = Arc::new(Persister::new(data_dir));
1736 let (exit_sender, exit_receiver) = tokio::sync::watch::channel(());
1738 let bp_future = super::process_events_async(
1739 persister, event_handler, nodes[0].chain_monitor.clone(), nodes[0].node.clone(),
1740 nodes[0].no_gossip_sync(), nodes[0].peer_manager.clone(), nodes[0].logger.clone(),
1741 Some(nodes[0].scorer.clone()), move |dur: Duration| {
1742 let mut exit_receiver = exit_receiver.clone();
1743 Box::pin(async move {
1745 _ = tokio::time::sleep(dur) => false,
1746 _ = exit_receiver.changed() => true,
1751 let t1 = tokio::spawn(bp_future);
1752 let t2 = tokio::spawn(async move {
1753 do_test_payment_path_scoring!(nodes, receiver.recv().await);
1754 exit_sender.send(()).unwrap();
1757 let (r1, r2) = tokio::join!(t1, t2);
1758 r1.unwrap().unwrap();