1 //! Utilities that take care of tasks that (1) need to happen periodically to keep Rust-Lightning
2 //! running properly, and (2) either can or should be run in the background. See docs for
3 //! [`BackgroundProcessor`] for more details on the nitty-gritty.
5 // Prefix these with `rustdoc::` when we update our MSRV to be >= 1.52 to remove warnings.
6 #![deny(broken_intra_doc_links)]
7 #![deny(private_intra_doc_links)]
10 #![cfg_attr(not(feature = "futures"), deny(unsafe_code))]
12 #![cfg_attr(docsrs, feature(doc_auto_cfg))]
14 #![cfg_attr(all(not(feature = "std"), not(test)), no_std)]
16 #[cfg(any(test, feature = "std"))]
19 #[cfg(not(feature = "std"))]
22 #[macro_use] extern crate lightning;
23 extern crate lightning_rapid_gossip_sync;
26 use lightning::chain::chaininterface::{BroadcasterInterface, FeeEstimator};
27 use lightning::chain::chainmonitor::{ChainMonitor, Persist};
28 use lightning::sign::{EntropySource, NodeSigner, SignerProvider};
29 use lightning::events::{Event, PathFailure};
30 #[cfg(feature = "std")]
31 use lightning::events::{EventHandler, EventsProvider};
32 use lightning::ln::channelmanager::ChannelManager;
33 use lightning::ln::peer_handler::APeerManager;
34 use lightning::routing::gossip::{NetworkGraph, P2PGossipSync};
35 use lightning::routing::utxo::UtxoLookup;
36 use lightning::routing::router::Router;
37 use lightning::routing::scoring::{Score, WriteableScore};
38 use lightning::util::logger::Logger;
39 use lightning::util::persist::Persister;
40 #[cfg(feature = "std")]
41 use lightning::util::wakers::Sleeper;
42 use lightning_rapid_gossip_sync::RapidGossipSync;
45 use core::time::Duration;
47 #[cfg(feature = "std")]
49 #[cfg(feature = "std")]
50 use core::sync::atomic::{AtomicBool, Ordering};
51 #[cfg(feature = "std")]
52 use std::thread::{self, JoinHandle};
53 #[cfg(feature = "std")]
54 use std::time::Instant;
56 #[cfg(not(feature = "std"))]
59 /// `BackgroundProcessor` takes care of tasks that (1) need to happen periodically to keep
60 /// Rust-Lightning running properly, and (2) either can or should be run in the background. Its
61 /// responsibilities are:
62 /// * Processing [`Event`]s with a user-provided [`EventHandler`].
63 /// * Monitoring whether the [`ChannelManager`] needs to be re-persisted to disk, and if so,
64 /// writing it to disk/backups by invoking the callback given to it at startup.
65 /// [`ChannelManager`] persistence should be done in the background.
66 /// * Calling [`ChannelManager::timer_tick_occurred`], [`ChainMonitor::rebroadcast_pending_claims`]
67 /// and [`PeerManager::timer_tick_occurred`] at the appropriate intervals.
68 /// * Calling [`NetworkGraph::remove_stale_channels_and_tracking`] (if a [`GossipSync`] with a
69 /// [`NetworkGraph`] is provided to [`BackgroundProcessor::start`]).
71 /// It will also call [`PeerManager::process_events`] periodically though this shouldn't be relied
72 /// upon as doing so may result in high latency.
76 /// If [`ChannelManager`] persistence fails and the persisted manager becomes out-of-date, then
77 /// there is a risk of channels force-closing on startup when the manager realizes it's outdated.
78 /// However, as long as [`ChannelMonitor`] backups are sound, no funds besides those used for
79 /// unilateral chain closure fees are at risk.
81 /// [`ChannelMonitor`]: lightning::chain::channelmonitor::ChannelMonitor
82 /// [`Event`]: lightning::events::Event
83 /// [`PeerManager::timer_tick_occurred`]: lightning::ln::peer_handler::PeerManager::timer_tick_occurred
84 /// [`PeerManager::process_events`]: lightning::ln::peer_handler::PeerManager::process_events
85 #[cfg(feature = "std")]
86 #[must_use = "BackgroundProcessor will immediately stop on drop. It should be stored until shutdown."]
87 pub struct BackgroundProcessor {
88 stop_thread: Arc<AtomicBool>,
89 thread_handle: Option<JoinHandle<Result<(), std::io::Error>>>,
93 const FRESHNESS_TIMER: u64 = 60;
95 const FRESHNESS_TIMER: u64 = 1;
97 #[cfg(all(not(test), not(debug_assertions)))]
98 const PING_TIMER: u64 = 10;
99 /// Signature operations take a lot longer without compiler optimisations.
100 /// Increasing the ping timer allows for this but slower devices will be disconnected if the
101 /// timeout is reached.
102 #[cfg(all(not(test), debug_assertions))]
103 const PING_TIMER: u64 = 30;
105 const PING_TIMER: u64 = 1;
107 /// Prune the network graph of stale entries hourly.
108 const NETWORK_PRUNE_TIMER: u64 = 60 * 60;
111 const SCORER_PERSIST_TIMER: u64 = 60 * 60;
113 const SCORER_PERSIST_TIMER: u64 = 1;
116 const FIRST_NETWORK_PRUNE_TIMER: u64 = 60;
118 const FIRST_NETWORK_PRUNE_TIMER: u64 = 1;
121 const REBROADCAST_TIMER: u64 = 30;
123 const REBROADCAST_TIMER: u64 = 1;
125 #[cfg(feature = "futures")]
126 /// core::cmp::min is not currently const, so we define a trivial (and equivalent) replacement
127 const fn min_u64(a: u64, b: u64) -> u64 { if a < b { a } else { b } }
128 #[cfg(feature = "futures")]
129 const FASTEST_TIMER: u64 = min_u64(min_u64(FRESHNESS_TIMER, PING_TIMER),
130 min_u64(SCORER_PERSIST_TIMER, min_u64(FIRST_NETWORK_PRUNE_TIMER, REBROADCAST_TIMER)));
132 /// Either [`P2PGossipSync`] or [`RapidGossipSync`].
134 P: Deref<Target = P2PGossipSync<G, U, L>>,
135 R: Deref<Target = RapidGossipSync<G, L>>,
136 G: Deref<Target = NetworkGraph<L>>,
140 where U::Target: UtxoLookup, L::Target: Logger {
141 /// Gossip sync via the lightning peer-to-peer network as defined by BOLT 7.
143 /// Rapid gossip sync from a trusted server.
150 P: Deref<Target = P2PGossipSync<G, U, L>>,
151 R: Deref<Target = RapidGossipSync<G, L>>,
152 G: Deref<Target = NetworkGraph<L>>,
155 > GossipSync<P, R, G, U, L>
156 where U::Target: UtxoLookup, L::Target: Logger {
157 fn network_graph(&self) -> Option<&G> {
159 GossipSync::P2P(gossip_sync) => Some(gossip_sync.network_graph()),
160 GossipSync::Rapid(gossip_sync) => Some(gossip_sync.network_graph()),
161 GossipSync::None => None,
165 fn prunable_network_graph(&self) -> Option<&G> {
167 GossipSync::P2P(gossip_sync) => Some(gossip_sync.network_graph()),
168 GossipSync::Rapid(gossip_sync) => {
169 if gossip_sync.is_initial_sync_complete() {
170 Some(gossip_sync.network_graph())
175 GossipSync::None => None,
180 /// This is not exported to bindings users as the bindings concretize everything and have constructors for us
181 impl<P: Deref<Target = P2PGossipSync<G, U, L>>, G: Deref<Target = NetworkGraph<L>>, U: Deref, L: Deref>
182 GossipSync<P, &RapidGossipSync<G, L>, G, U, L>
184 U::Target: UtxoLookup,
187 /// Initializes a new [`GossipSync::P2P`] variant.
188 pub fn p2p(gossip_sync: P) -> Self {
189 GossipSync::P2P(gossip_sync)
193 /// This is not exported to bindings users as the bindings concretize everything and have constructors for us
194 impl<'a, R: Deref<Target = RapidGossipSync<G, L>>, G: Deref<Target = NetworkGraph<L>>, L: Deref>
196 &P2PGossipSync<G, &'a (dyn UtxoLookup + Send + Sync), L>,
199 &'a (dyn UtxoLookup + Send + Sync),
205 /// Initializes a new [`GossipSync::Rapid`] variant.
206 pub fn rapid(gossip_sync: R) -> Self {
207 GossipSync::Rapid(gossip_sync)
211 /// This is not exported to bindings users as the bindings concretize everything and have constructors for us
214 &P2PGossipSync<&'a NetworkGraph<L>, &'a (dyn UtxoLookup + Send + Sync), L>,
215 &RapidGossipSync<&'a NetworkGraph<L>, L>,
217 &'a (dyn UtxoLookup + Send + Sync),
223 /// Initializes a new [`GossipSync::None`] variant.
224 pub fn none() -> Self {
229 fn handle_network_graph_update<L: Deref>(
230 network_graph: &NetworkGraph<L>, event: &Event
231 ) where L::Target: Logger {
232 if let Event::PaymentPathFailed {
233 failure: PathFailure::OnPath { network_update: Some(ref upd) }, .. } = event
235 network_graph.handle_network_update(upd);
239 /// Updates scorer based on event and returns whether an update occurred so we can decide whether
241 fn update_scorer<'a, S: 'static + Deref<Target = SC> + Send + Sync, SC: 'a + WriteableScore<'a>>(
242 scorer: &'a S, event: &Event
244 let mut score = scorer.lock();
246 Event::PaymentPathFailed { ref path, short_channel_id: Some(scid), .. } => {
247 score.payment_path_failed(path, *scid);
249 Event::PaymentPathFailed { ref path, payment_failed_permanently: true, .. } => {
250 // Reached if the destination explicitly failed it back. We treat this as a successful probe
251 // because the payment made it all the way to the destination with sufficient liquidity.
252 score.probe_successful(path);
254 Event::PaymentPathSuccessful { path, .. } => {
255 score.payment_path_successful(path);
257 Event::ProbeSuccessful { path, .. } => {
258 score.probe_successful(path);
260 Event::ProbeFailed { path, short_channel_id: Some(scid), .. } => {
261 score.probe_failed(path, *scid);
268 macro_rules! define_run_body {
269 ($persister: ident, $chain_monitor: ident, $process_chain_monitor_events: expr,
270 $channel_manager: ident, $process_channel_manager_events: expr,
271 $gossip_sync: ident, $peer_manager: ident, $logger: ident, $scorer: ident,
272 $loop_exit_check: expr, $await: expr, $get_timer: expr, $timer_elapsed: expr,
273 $check_slow_await: expr)
275 log_trace!($logger, "Calling ChannelManager's timer_tick_occurred on startup");
276 $channel_manager.timer_tick_occurred();
277 log_trace!($logger, "Rebroadcasting monitor's pending claims on startup");
278 $chain_monitor.rebroadcast_pending_claims();
280 let mut last_freshness_call = $get_timer(FRESHNESS_TIMER);
281 let mut last_ping_call = $get_timer(PING_TIMER);
282 let mut last_prune_call = $get_timer(FIRST_NETWORK_PRUNE_TIMER);
283 let mut last_scorer_persist_call = $get_timer(SCORER_PERSIST_TIMER);
284 let mut last_rebroadcast_call = $get_timer(REBROADCAST_TIMER);
285 let mut have_pruned = false;
288 $process_channel_manager_events;
289 $process_chain_monitor_events;
291 // Note that the PeerManager::process_events may block on ChannelManager's locks,
292 // hence it comes last here. When the ChannelManager finishes whatever it's doing,
293 // we want to ensure we get into `persist_manager` as quickly as we can, especially
294 // without running the normal event processing above and handing events to users.
296 // Specifically, on an *extremely* slow machine, we may see ChannelManager start
297 // processing a message effectively at any point during this loop. In order to
298 // minimize the time between such processing completing and persisting the updated
299 // ChannelManager, we want to minimize methods blocking on a ChannelManager
300 // generally, and as a fallback place such blocking only immediately before
302 $peer_manager.as_ref().process_events();
304 // Exit the loop if the background processor was requested to stop.
305 if $loop_exit_check {
306 log_trace!($logger, "Terminating background processor.");
310 // We wait up to 100ms, but track how long it takes to detect being put to sleep,
311 // see `await_start`'s use below.
312 let mut await_start = None;
313 if $check_slow_await { await_start = Some($get_timer(1)); }
314 let updates_available = $await;
315 let await_slow = if $check_slow_await { $timer_elapsed(&mut await_start.unwrap(), 1) } else { false };
317 // Exit the loop if the background processor was requested to stop.
318 if $loop_exit_check {
319 log_trace!($logger, "Terminating background processor.");
323 if updates_available {
324 log_trace!($logger, "Persisting ChannelManager...");
325 $persister.persist_manager(&*$channel_manager)?;
326 log_trace!($logger, "Done persisting ChannelManager.");
328 if $timer_elapsed(&mut last_freshness_call, FRESHNESS_TIMER) {
329 log_trace!($logger, "Calling ChannelManager's timer_tick_occurred");
330 $channel_manager.timer_tick_occurred();
331 last_freshness_call = $get_timer(FRESHNESS_TIMER);
334 // On various platforms, we may be starved of CPU cycles for several reasons.
335 // E.g. on iOS, if we've been in the background, we will be entirely paused.
336 // Similarly, if we're on a desktop platform and the device has been asleep, we
337 // may not get any cycles.
338 // We detect this by checking if our max-100ms-sleep, above, ran longer than a
339 // full second, at which point we assume sockets may have been killed (they
340 // appear to be at least on some platforms, even if it has only been a second).
341 // Note that we have to take care to not get here just because user event
342 // processing was slow at the top of the loop. For example, the sample client
343 // may call Bitcoin Core RPCs during event handling, which very often takes
344 // more than a handful of seconds to complete, and shouldn't disconnect all our
346 log_trace!($logger, "100ms sleep took more than a second, disconnecting peers.");
347 $peer_manager.as_ref().disconnect_all_peers();
348 last_ping_call = $get_timer(PING_TIMER);
349 } else if $timer_elapsed(&mut last_ping_call, PING_TIMER) {
350 log_trace!($logger, "Calling PeerManager's timer_tick_occurred");
351 $peer_manager.as_ref().timer_tick_occurred();
352 last_ping_call = $get_timer(PING_TIMER);
355 // Note that we want to run a graph prune once not long after startup before
356 // falling back to our usual hourly prunes. This avoids short-lived clients never
357 // pruning their network graph. We run once 60 seconds after startup before
358 // continuing our normal cadence. For RGS, since 60 seconds is likely too long,
359 // we prune after an initial sync completes.
360 let prune_timer = if have_pruned { NETWORK_PRUNE_TIMER } else { FIRST_NETWORK_PRUNE_TIMER };
361 let prune_timer_elapsed = $timer_elapsed(&mut last_prune_call, prune_timer);
362 let should_prune = match $gossip_sync {
363 GossipSync::Rapid(_) => !have_pruned || prune_timer_elapsed,
364 _ => prune_timer_elapsed,
367 // The network graph must not be pruned while rapid sync completion is pending
368 if let Some(network_graph) = $gossip_sync.prunable_network_graph() {
369 #[cfg(feature = "std")] {
370 log_trace!($logger, "Pruning and persisting network graph.");
371 network_graph.remove_stale_channels_and_tracking();
373 #[cfg(not(feature = "std"))] {
374 log_warn!($logger, "Not pruning network graph, consider enabling `std` or doing so manually with remove_stale_channels_and_tracking_with_time.");
375 log_trace!($logger, "Persisting network graph.");
378 if let Err(e) = $persister.persist_graph(network_graph) {
379 log_error!($logger, "Error: Failed to persist network graph, check your disk and permissions {}", e)
384 let prune_timer = if have_pruned { NETWORK_PRUNE_TIMER } else { FIRST_NETWORK_PRUNE_TIMER };
385 last_prune_call = $get_timer(prune_timer);
388 if $timer_elapsed(&mut last_scorer_persist_call, SCORER_PERSIST_TIMER) {
389 if let Some(ref scorer) = $scorer {
390 log_trace!($logger, "Persisting scorer");
391 if let Err(e) = $persister.persist_scorer(&scorer) {
392 log_error!($logger, "Error: Failed to persist scorer, check your disk and permissions {}", e)
395 last_scorer_persist_call = $get_timer(SCORER_PERSIST_TIMER);
398 if $timer_elapsed(&mut last_rebroadcast_call, REBROADCAST_TIMER) {
399 log_trace!($logger, "Rebroadcasting monitor's pending claims");
400 $chain_monitor.rebroadcast_pending_claims();
401 last_rebroadcast_call = $get_timer(REBROADCAST_TIMER);
405 // After we exit, ensure we persist the ChannelManager one final time - this avoids
406 // some races where users quit while channel updates were in-flight, with
407 // ChannelMonitor update(s) persisted without a corresponding ChannelManager update.
408 $persister.persist_manager(&*$channel_manager)?;
410 // Persist Scorer on exit
411 if let Some(ref scorer) = $scorer {
412 $persister.persist_scorer(&scorer)?;
415 // Persist NetworkGraph on exit
416 if let Some(network_graph) = $gossip_sync.network_graph() {
417 $persister.persist_graph(network_graph)?;
424 #[cfg(feature = "futures")]
425 pub(crate) mod futures_util {
426 use core::future::Future;
427 use core::task::{Poll, Waker, RawWaker, RawWakerVTable};
429 use core::marker::Unpin;
430 pub(crate) struct Selector<
431 A: Future<Output=()> + Unpin, B: Future<Output=()> + Unpin, C: Future<Output=bool> + Unpin
437 pub(crate) enum SelectorOutput {
442 A: Future<Output=()> + Unpin, B: Future<Output=()> + Unpin, C: Future<Output=bool> + Unpin
443 > Future for Selector<A, B, C> {
444 type Output = SelectorOutput;
445 fn poll(mut self: Pin<&mut Self>, ctx: &mut core::task::Context<'_>) -> Poll<SelectorOutput> {
446 match Pin::new(&mut self.a).poll(ctx) {
447 Poll::Ready(()) => { return Poll::Ready(SelectorOutput::A); },
450 match Pin::new(&mut self.b).poll(ctx) {
451 Poll::Ready(()) => { return Poll::Ready(SelectorOutput::B); },
454 match Pin::new(&mut self.c).poll(ctx) {
455 Poll::Ready(res) => { return Poll::Ready(SelectorOutput::C(res)); },
462 // If we want to poll a future without an async context to figure out if it has completed or
463 // not without awaiting, we need a Waker, which needs a vtable...we fill it with dummy values
464 // but sadly there's a good bit of boilerplate here.
465 fn dummy_waker_clone(_: *const ()) -> RawWaker { RawWaker::new(core::ptr::null(), &DUMMY_WAKER_VTABLE) }
466 fn dummy_waker_action(_: *const ()) { }
468 const DUMMY_WAKER_VTABLE: RawWakerVTable = RawWakerVTable::new(
469 dummy_waker_clone, dummy_waker_action, dummy_waker_action, dummy_waker_action);
470 pub(crate) fn dummy_waker() -> Waker { unsafe { Waker::from_raw(RawWaker::new(core::ptr::null(), &DUMMY_WAKER_VTABLE)) } }
472 #[cfg(feature = "futures")]
473 use futures_util::{Selector, SelectorOutput, dummy_waker};
474 #[cfg(feature = "futures")]
477 /// Processes background events in a future.
479 /// `sleeper` should return a future which completes in the given amount of time and returns a
480 /// boolean indicating whether the background processing should exit. Once `sleeper` returns a
481 /// future which outputs `true`, the loop will exit and this function's future will complete.
482 /// The `sleeper` future is free to return early after it has triggered the exit condition.
484 /// See [`BackgroundProcessor::start`] for information on which actions this handles.
486 /// Requires the `futures` feature. Note that while this method is available without the `std`
487 /// feature, doing so will skip calling [`NetworkGraph::remove_stale_channels_and_tracking`],
488 /// you should call [`NetworkGraph::remove_stale_channels_and_tracking_with_time`] regularly
489 /// manually instead.
491 /// The `mobile_interruptable_platform` flag should be set if we're currently running on a
492 /// mobile device, where we may need to check for interruption of the application regularly. If you
493 /// are unsure, you should set the flag, as the performance impact of it is minimal unless there
494 /// are hundreds or thousands of simultaneous process calls running.
496 /// For example, in order to process background events in a [Tokio](https://tokio.rs/) task, you
497 /// could setup `process_events_async` like this:
499 /// # use lightning::io;
500 /// # use std::sync::{Arc, Mutex};
501 /// # use std::sync::atomic::{AtomicBool, Ordering};
502 /// # use lightning_background_processor::{process_events_async, GossipSync};
503 /// # struct MyStore {}
504 /// # impl lightning::util::persist::KVStore for MyStore {
505 /// # type Reader = io::Cursor<Vec<u8>>;
506 /// # fn read(&self, namespace: &str, key: &str) -> io::Result<Self::Reader> { Ok(io::Cursor::new(Vec::new())) }
507 /// # fn write(&self, namespace: &str, key: &str, buf: &[u8]) -> io::Result<()> { Ok(()) }
508 /// # fn remove(&self, namespace: &str, key: &str) -> io::Result<()> { Ok(()) }
509 /// # fn list(&self, namespace: &str) -> io::Result<Vec<String>> { Ok(Vec::new()) }
511 /// # struct MyEventHandler {}
512 /// # impl MyEventHandler {
513 /// # async fn handle_event(&self, _: lightning::events::Event) {}
515 /// # #[derive(Eq, PartialEq, Clone, Hash)]
516 /// # struct MySocketDescriptor {}
517 /// # impl lightning::ln::peer_handler::SocketDescriptor for MySocketDescriptor {
518 /// # fn send_data(&mut self, _data: &[u8], _resume_read: bool) -> usize { 0 }
519 /// # fn disconnect_socket(&mut self) {}
521 /// # type MyBroadcaster = dyn lightning::chain::chaininterface::BroadcasterInterface + Send + Sync;
522 /// # type MyFeeEstimator = dyn lightning::chain::chaininterface::FeeEstimator + Send + Sync;
523 /// # type MyNodeSigner = dyn lightning::sign::NodeSigner + Send + Sync;
524 /// # type MyUtxoLookup = dyn lightning::routing::utxo::UtxoLookup + Send + Sync;
525 /// # type MyFilter = dyn lightning::chain::Filter + Send + Sync;
526 /// # type MyLogger = dyn lightning::util::logger::Logger + Send + Sync;
527 /// # type MyChainMonitor = lightning::chain::chainmonitor::ChainMonitor<lightning::sign::InMemorySigner, Arc<MyFilter>, Arc<MyBroadcaster>, Arc<MyFeeEstimator>, Arc<MyLogger>, Arc<MyStore>>;
528 /// # type MyPeerManager = lightning::ln::peer_handler::SimpleArcPeerManager<MySocketDescriptor, MyChainMonitor, MyBroadcaster, MyFeeEstimator, MyUtxoLookup, MyLogger>;
529 /// # type MyNetworkGraph = lightning::routing::gossip::NetworkGraph<Arc<MyLogger>>;
530 /// # type MyGossipSync = lightning::routing::gossip::P2PGossipSync<Arc<MyNetworkGraph>, Arc<MyUtxoLookup>, Arc<MyLogger>>;
531 /// # type MyChannelManager = lightning::ln::channelmanager::SimpleArcChannelManager<MyChainMonitor, MyBroadcaster, MyFeeEstimator, MyLogger>;
532 /// # type MyScorer = Mutex<lightning::routing::scoring::ProbabilisticScorer<Arc<MyNetworkGraph>, Arc<MyLogger>>>;
534 /// # 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>) {
535 /// let background_persister = Arc::clone(&my_persister);
536 /// let background_event_handler = Arc::clone(&my_event_handler);
537 /// let background_chain_mon = Arc::clone(&my_chain_monitor);
538 /// let background_chan_man = Arc::clone(&my_channel_manager);
539 /// let background_gossip_sync = GossipSync::p2p(Arc::clone(&my_gossip_sync));
540 /// let background_peer_man = Arc::clone(&my_peer_manager);
541 /// let background_logger = Arc::clone(&my_logger);
542 /// let background_scorer = Arc::clone(&my_scorer);
544 /// // Setup the sleeper.
545 /// let (stop_sender, stop_receiver) = tokio::sync::watch::channel(());
547 /// let sleeper = move |d| {
548 /// let mut receiver = stop_receiver.clone();
549 /// Box::pin(async move {
551 /// _ = tokio::time::sleep(d) => false,
552 /// _ = receiver.changed() => true,
557 /// let mobile_interruptable_platform = false;
559 /// let handle = tokio::spawn(async move {
560 /// process_events_async(
561 /// background_persister,
562 /// |e| background_event_handler.handle_event(e),
563 /// background_chain_mon,
564 /// background_chan_man,
565 /// background_gossip_sync,
566 /// background_peer_man,
567 /// background_logger,
568 /// Some(background_scorer),
570 /// mobile_interruptable_platform,
573 /// .expect("Failed to process events");
576 /// // Stop the background processing.
577 /// stop_sender.send(()).unwrap();
578 /// handle.await.unwrap();
581 #[cfg(feature = "futures")]
582 pub async fn process_events_async<
584 UL: 'static + Deref + Send + Sync,
585 CF: 'static + Deref + Send + Sync,
586 CW: 'static + Deref + Send + Sync,
587 T: 'static + Deref + Send + Sync,
588 ES: 'static + Deref + Send + Sync,
589 NS: 'static + Deref + Send + Sync,
590 SP: 'static + Deref + Send + Sync,
591 F: 'static + Deref + Send + Sync,
592 R: 'static + Deref + Send + Sync,
593 G: 'static + Deref<Target = NetworkGraph<L>> + Send + Sync,
594 L: 'static + Deref + Send + Sync,
595 P: 'static + Deref + Send + Sync,
596 EventHandlerFuture: core::future::Future<Output = ()>,
597 EventHandler: Fn(Event) -> EventHandlerFuture,
598 PS: 'static + Deref + Send,
599 M: 'static + Deref<Target = ChainMonitor<<SP::Target as SignerProvider>::Signer, CF, T, F, L, P>> + Send + Sync,
600 CM: 'static + Deref<Target = ChannelManager<CW, T, ES, NS, SP, F, R, L>> + Send + Sync,
601 PGS: 'static + Deref<Target = P2PGossipSync<G, UL, L>> + Send + Sync,
602 RGS: 'static + Deref<Target = RapidGossipSync<G, L>> + Send,
603 APM: APeerManager + Send + Sync,
604 PM: 'static + Deref<Target = APM> + Send + Sync,
605 S: 'static + Deref<Target = SC> + Send + Sync,
606 SC: for<'b> WriteableScore<'b>,
607 SleepFuture: core::future::Future<Output = bool> + core::marker::Unpin,
608 Sleeper: Fn(Duration) -> SleepFuture
610 persister: PS, event_handler: EventHandler, chain_monitor: M, channel_manager: CM,
611 gossip_sync: GossipSync<PGS, RGS, G, UL, L>, peer_manager: PM, logger: L, scorer: Option<S>,
612 sleeper: Sleeper, mobile_interruptable_platform: bool,
613 ) -> Result<(), lightning::io::Error>
615 UL::Target: 'static + UtxoLookup,
616 CF::Target: 'static + chain::Filter,
617 CW::Target: 'static + chain::Watch<<SP::Target as SignerProvider>::Signer>,
618 T::Target: 'static + BroadcasterInterface,
619 ES::Target: 'static + EntropySource,
620 NS::Target: 'static + NodeSigner,
621 SP::Target: 'static + SignerProvider,
622 F::Target: 'static + FeeEstimator,
623 R::Target: 'static + Router,
624 L::Target: 'static + Logger,
625 P::Target: 'static + Persist<<SP::Target as SignerProvider>::Signer>,
626 PS::Target: 'static + Persister<'a, CW, T, ES, NS, SP, F, R, L, SC>,
628 let mut should_break = false;
629 let async_event_handler = |event| {
630 let network_graph = gossip_sync.network_graph();
631 let event_handler = &event_handler;
632 let scorer = &scorer;
633 let logger = &logger;
634 let persister = &persister;
636 if let Some(network_graph) = network_graph {
637 handle_network_graph_update(network_graph, &event)
639 if let Some(ref scorer) = scorer {
640 if update_scorer(scorer, &event) {
641 log_trace!(logger, "Persisting scorer after update");
642 if let Err(e) = persister.persist_scorer(&scorer) {
643 log_error!(logger, "Error: Failed to persist scorer, check your disk and permissions {}", e)
647 event_handler(event).await;
650 define_run_body!(persister,
651 chain_monitor, chain_monitor.process_pending_events_async(async_event_handler).await,
652 channel_manager, channel_manager.process_pending_events_async(async_event_handler).await,
653 gossip_sync, peer_manager, logger, scorer, should_break, {
655 a: channel_manager.get_persistable_update_future(),
656 b: chain_monitor.get_update_future(),
657 c: sleeper(if mobile_interruptable_platform { Duration::from_millis(100) } else { Duration::from_secs(FASTEST_TIMER) }),
660 SelectorOutput::A => true,
661 SelectorOutput::B => false,
662 SelectorOutput::C(exit) => {
667 }, |t| sleeper(Duration::from_secs(t)),
668 |fut: &mut SleepFuture, _| {
669 let mut waker = dummy_waker();
670 let mut ctx = task::Context::from_waker(&mut waker);
671 match core::pin::Pin::new(fut).poll(&mut ctx) {
672 task::Poll::Ready(exit) => { should_break = exit; true },
673 task::Poll::Pending => false,
675 }, mobile_interruptable_platform)
678 #[cfg(feature = "std")]
679 impl BackgroundProcessor {
680 /// Start a background thread that takes care of responsibilities enumerated in the [top-level
683 /// The thread runs indefinitely unless the object is dropped, [`stop`] is called, or
684 /// [`Persister::persist_manager`] returns an error. In case of an error, the error is retrieved by calling
685 /// either [`join`] or [`stop`].
687 /// # Data Persistence
689 /// [`Persister::persist_manager`] is responsible for writing out the [`ChannelManager`] to disk, and/or
690 /// uploading to one or more backup services. See [`ChannelManager::write`] for writing out a
691 /// [`ChannelManager`]. See the `lightning-persister` crate for LDK's
692 /// provided implementation.
694 /// [`Persister::persist_graph`] is responsible for writing out the [`NetworkGraph`] to disk, if
695 /// [`GossipSync`] is supplied. See [`NetworkGraph::write`] for writing out a [`NetworkGraph`].
696 /// See the `lightning-persister` crate for LDK's provided implementation.
698 /// Typically, users should either implement [`Persister::persist_manager`] to never return an
699 /// error or call [`join`] and handle any error that may arise. For the latter case,
700 /// `BackgroundProcessor` must be restarted by calling `start` again after handling the error.
704 /// `event_handler` is responsible for handling events that users should be notified of (e.g.,
705 /// payment failed). [`BackgroundProcessor`] may decorate the given [`EventHandler`] with common
706 /// functionality implemented by other handlers.
707 /// * [`P2PGossipSync`] if given will update the [`NetworkGraph`] based on payment failures.
709 /// # Rapid Gossip Sync
711 /// If rapid gossip sync is meant to run at startup, pass [`RapidGossipSync`] via `gossip_sync`
712 /// to indicate that the [`BackgroundProcessor`] should not prune the [`NetworkGraph`] instance
713 /// until the [`RapidGossipSync`] instance completes its first sync.
715 /// [top-level documentation]: BackgroundProcessor
716 /// [`join`]: Self::join
717 /// [`stop`]: Self::stop
718 /// [`ChannelManager`]: lightning::ln::channelmanager::ChannelManager
719 /// [`ChannelManager::write`]: lightning::ln::channelmanager::ChannelManager#impl-Writeable
720 /// [`Persister::persist_manager`]: lightning::util::persist::Persister::persist_manager
721 /// [`Persister::persist_graph`]: lightning::util::persist::Persister::persist_graph
722 /// [`NetworkGraph`]: lightning::routing::gossip::NetworkGraph
723 /// [`NetworkGraph::write`]: lightning::routing::gossip::NetworkGraph#impl-Writeable
726 UL: 'static + Deref + Send + Sync,
727 CF: 'static + Deref + Send + Sync,
728 CW: 'static + Deref + Send + Sync,
729 T: 'static + Deref + Send + Sync,
730 ES: 'static + Deref + Send + Sync,
731 NS: 'static + Deref + Send + Sync,
732 SP: 'static + Deref + Send + Sync,
733 F: 'static + Deref + Send + Sync,
734 R: 'static + Deref + Send + Sync,
735 G: 'static + Deref<Target = NetworkGraph<L>> + Send + Sync,
736 L: 'static + Deref + Send + Sync,
737 P: 'static + Deref + Send + Sync,
738 EH: 'static + EventHandler + Send,
739 PS: 'static + Deref + Send,
740 M: 'static + Deref<Target = ChainMonitor<<SP::Target as SignerProvider>::Signer, CF, T, F, L, P>> + Send + Sync,
741 CM: 'static + Deref<Target = ChannelManager<CW, T, ES, NS, SP, F, R, L>> + Send + Sync,
742 PGS: 'static + Deref<Target = P2PGossipSync<G, UL, L>> + Send + Sync,
743 RGS: 'static + Deref<Target = RapidGossipSync<G, L>> + Send,
744 APM: APeerManager + Send + Sync,
745 PM: 'static + Deref<Target = APM> + Send + Sync,
746 S: 'static + Deref<Target = SC> + Send + Sync,
747 SC: for <'b> WriteableScore<'b>,
749 persister: PS, event_handler: EH, chain_monitor: M, channel_manager: CM,
750 gossip_sync: GossipSync<PGS, RGS, G, UL, L>, peer_manager: PM, logger: L, scorer: Option<S>,
753 UL::Target: 'static + UtxoLookup,
754 CF::Target: 'static + chain::Filter,
755 CW::Target: 'static + chain::Watch<<SP::Target as SignerProvider>::Signer>,
756 T::Target: 'static + BroadcasterInterface,
757 ES::Target: 'static + EntropySource,
758 NS::Target: 'static + NodeSigner,
759 SP::Target: 'static + SignerProvider,
760 F::Target: 'static + FeeEstimator,
761 R::Target: 'static + Router,
762 L::Target: 'static + Logger,
763 P::Target: 'static + Persist<<SP::Target as SignerProvider>::Signer>,
764 PS::Target: 'static + Persister<'a, CW, T, ES, NS, SP, F, R, L, SC>,
766 let stop_thread = Arc::new(AtomicBool::new(false));
767 let stop_thread_clone = stop_thread.clone();
768 let handle = thread::spawn(move || -> Result<(), std::io::Error> {
769 let event_handler = |event| {
770 let network_graph = gossip_sync.network_graph();
771 if let Some(network_graph) = network_graph {
772 handle_network_graph_update(network_graph, &event)
774 if let Some(ref scorer) = scorer {
775 if update_scorer(scorer, &event) {
776 log_trace!(logger, "Persisting scorer after update");
777 if let Err(e) = persister.persist_scorer(&scorer) {
778 log_error!(logger, "Error: Failed to persist scorer, check your disk and permissions {}", e)
782 event_handler.handle_event(event);
784 define_run_body!(persister, chain_monitor, chain_monitor.process_pending_events(&event_handler),
785 channel_manager, channel_manager.process_pending_events(&event_handler),
786 gossip_sync, peer_manager, logger, scorer, stop_thread.load(Ordering::Acquire),
787 Sleeper::from_two_futures(
788 channel_manager.get_persistable_update_future(),
789 chain_monitor.get_update_future()
790 ).wait_timeout(Duration::from_millis(100)),
791 |_| Instant::now(), |time: &Instant, dur| time.elapsed().as_secs() > dur, false)
793 Self { stop_thread: stop_thread_clone, thread_handle: Some(handle) }
796 /// Join `BackgroundProcessor`'s thread, returning any error that occurred while persisting
797 /// [`ChannelManager`].
801 /// This function panics if the background thread has panicked such as while persisting or
804 /// [`ChannelManager`]: lightning::ln::channelmanager::ChannelManager
805 pub fn join(mut self) -> Result<(), std::io::Error> {
806 assert!(self.thread_handle.is_some());
810 /// Stop `BackgroundProcessor`'s thread, returning any error that occurred while persisting
811 /// [`ChannelManager`].
815 /// This function panics if the background thread has panicked such as while persisting or
818 /// [`ChannelManager`]: lightning::ln::channelmanager::ChannelManager
819 pub fn stop(mut self) -> Result<(), std::io::Error> {
820 assert!(self.thread_handle.is_some());
821 self.stop_and_join_thread()
824 fn stop_and_join_thread(&mut self) -> Result<(), std::io::Error> {
825 self.stop_thread.store(true, Ordering::Release);
829 fn join_thread(&mut self) -> Result<(), std::io::Error> {
830 match self.thread_handle.take() {
831 Some(handle) => handle.join().unwrap(),
837 #[cfg(feature = "std")]
838 impl Drop for BackgroundProcessor {
840 self.stop_and_join_thread().unwrap();
844 #[cfg(all(feature = "std", test))]
846 use bitcoin::blockdata::constants::{genesis_block, ChainHash};
847 use bitcoin::blockdata::locktime::PackedLockTime;
848 use bitcoin::blockdata::transaction::{Transaction, TxOut};
849 use bitcoin::network::constants::Network;
850 use bitcoin::secp256k1::{SecretKey, PublicKey, Secp256k1};
851 use lightning::chain::{BestBlock, Confirm, chainmonitor};
852 use lightning::chain::channelmonitor::ANTI_REORG_DELAY;
853 use lightning::sign::{InMemorySigner, KeysManager};
854 use lightning::chain::transaction::OutPoint;
855 use lightning::events::{Event, PathFailure, MessageSendEventsProvider, MessageSendEvent};
856 use lightning::{get_event_msg, get_event};
857 use lightning::ln::PaymentHash;
858 use lightning::ln::channelmanager;
859 use lightning::ln::channelmanager::{BREAKDOWN_TIMEOUT, ChainParameters, MIN_CLTV_EXPIRY_DELTA, PaymentId};
860 use lightning::ln::features::{ChannelFeatures, NodeFeatures};
861 use lightning::ln::functional_test_utils::*;
862 use lightning::ln::msgs::{ChannelMessageHandler, Init};
863 use lightning::ln::peer_handler::{PeerManager, MessageHandler, SocketDescriptor, IgnoringMessageHandler};
864 use lightning::routing::gossip::{NetworkGraph, NodeId, P2PGossipSync};
865 use lightning::routing::router::{DefaultRouter, Path, RouteHop};
866 use lightning::routing::scoring::{ChannelUsage, Score};
867 use lightning::util::config::UserConfig;
868 use lightning::util::ser::Writeable;
869 use lightning::util::test_utils;
870 use lightning::util::persist::{KVStore, CHANNEL_MANAGER_PERSISTENCE_NAMESPACE, CHANNEL_MANAGER_PERSISTENCE_KEY, NETWORK_GRAPH_PERSISTENCE_NAMESPACE, NETWORK_GRAPH_PERSISTENCE_KEY, SCORER_PERSISTENCE_NAMESPACE, SCORER_PERSISTENCE_KEY};
871 use lightning_persister::fs_store::FilesystemStore;
872 use std::collections::VecDeque;
874 use std::path::PathBuf;
875 use std::sync::{Arc, Mutex};
876 use std::sync::mpsc::SyncSender;
877 use std::time::Duration;
878 use lightning_rapid_gossip_sync::RapidGossipSync;
879 use super::{BackgroundProcessor, GossipSync, FRESHNESS_TIMER};
881 const EVENT_DEADLINE: u64 = 5 * FRESHNESS_TIMER;
883 #[derive(Clone, Hash, PartialEq, Eq)]
884 struct TestDescriptor{}
885 impl SocketDescriptor for TestDescriptor {
886 fn send_data(&mut self, _data: &[u8], _resume_read: bool) -> usize {
890 fn disconnect_socket(&mut self) {}
893 type ChannelManager =
894 channelmanager::ChannelManager<
896 Arc<test_utils::TestBroadcaster>,
900 Arc<test_utils::TestFeeEstimator>,
902 Arc<NetworkGraph<Arc<test_utils::TestLogger>>>,
903 Arc<test_utils::TestLogger>,
904 Arc<Mutex<TestScorer>>,
908 Arc<test_utils::TestLogger>>;
910 type ChainMonitor = chainmonitor::ChainMonitor<InMemorySigner, Arc<test_utils::TestChainSource>, Arc<test_utils::TestBroadcaster>, Arc<test_utils::TestFeeEstimator>, Arc<test_utils::TestLogger>, Arc<FilesystemStore>>;
912 type PGS = Arc<P2PGossipSync<Arc<NetworkGraph<Arc<test_utils::TestLogger>>>, Arc<test_utils::TestChainSource>, Arc<test_utils::TestLogger>>>;
913 type RGS = Arc<RapidGossipSync<Arc<NetworkGraph<Arc<test_utils::TestLogger>>>, Arc<test_utils::TestLogger>>>;
916 node: Arc<ChannelManager>,
917 p2p_gossip_sync: PGS,
918 rapid_gossip_sync: RGS,
919 peer_manager: Arc<PeerManager<TestDescriptor, Arc<test_utils::TestChannelMessageHandler>, Arc<test_utils::TestRoutingMessageHandler>, IgnoringMessageHandler, Arc<test_utils::TestLogger>, IgnoringMessageHandler, Arc<KeysManager>>>,
920 chain_monitor: Arc<ChainMonitor>,
921 kv_store: Arc<FilesystemStore>,
922 tx_broadcaster: Arc<test_utils::TestBroadcaster>,
923 network_graph: Arc<NetworkGraph<Arc<test_utils::TestLogger>>>,
924 logger: Arc<test_utils::TestLogger>,
925 best_block: BestBlock,
926 scorer: Arc<Mutex<TestScorer>>,
930 fn p2p_gossip_sync(&self) -> GossipSync<PGS, RGS, Arc<NetworkGraph<Arc<test_utils::TestLogger>>>, Arc<test_utils::TestChainSource>, Arc<test_utils::TestLogger>> {
931 GossipSync::P2P(self.p2p_gossip_sync.clone())
934 fn rapid_gossip_sync(&self) -> GossipSync<PGS, RGS, Arc<NetworkGraph<Arc<test_utils::TestLogger>>>, Arc<test_utils::TestChainSource>, Arc<test_utils::TestLogger>> {
935 GossipSync::Rapid(self.rapid_gossip_sync.clone())
938 fn no_gossip_sync(&self) -> GossipSync<PGS, RGS, Arc<NetworkGraph<Arc<test_utils::TestLogger>>>, Arc<test_utils::TestChainSource>, Arc<test_utils::TestLogger>> {
945 let data_dir = self.kv_store.get_data_dir();
946 match fs::remove_dir_all(data_dir.clone()) {
947 Err(e) => println!("Failed to remove test persister directory {}: {}", data_dir.display(), e),
954 graph_error: Option<(std::io::ErrorKind, &'static str)>,
955 graph_persistence_notifier: Option<SyncSender<()>>,
956 manager_error: Option<(std::io::ErrorKind, &'static str)>,
957 scorer_error: Option<(std::io::ErrorKind, &'static str)>,
958 kv_store: FilesystemStore,
962 fn new(data_dir: PathBuf) -> Self {
963 let kv_store = FilesystemStore::new(data_dir);
964 Self { graph_error: None, graph_persistence_notifier: None, manager_error: None, scorer_error: None, kv_store }
967 fn with_graph_error(self, error: std::io::ErrorKind, message: &'static str) -> Self {
968 Self { graph_error: Some((error, message)), ..self }
971 fn with_graph_persistence_notifier(self, sender: SyncSender<()>) -> Self {
972 Self { graph_persistence_notifier: Some(sender), ..self }
975 fn with_manager_error(self, error: std::io::ErrorKind, message: &'static str) -> Self {
976 Self { manager_error: Some((error, message)), ..self }
979 fn with_scorer_error(self, error: std::io::ErrorKind, message: &'static str) -> Self {
980 Self { scorer_error: Some((error, message)), ..self }
984 impl KVStore for Persister {
985 fn read(&self, namespace: &str, key: &str) -> lightning::io::Result<Vec<u8>> {
986 self.kv_store.read(namespace, key)
989 fn write(&self, namespace: &str, key: &str, buf: &[u8]) -> lightning::io::Result<()> {
990 if namespace == CHANNEL_MANAGER_PERSISTENCE_NAMESPACE && key == CHANNEL_MANAGER_PERSISTENCE_KEY {
991 if let Some((error, message)) = self.manager_error {
992 return Err(std::io::Error::new(error, message))
996 if namespace == NETWORK_GRAPH_PERSISTENCE_NAMESPACE && key == NETWORK_GRAPH_PERSISTENCE_KEY {
997 if let Some(sender) = &self.graph_persistence_notifier {
998 match sender.send(()) {
1000 Err(std::sync::mpsc::SendError(())) => println!("Persister failed to notify as receiver went away."),
1004 if let Some((error, message)) = self.graph_error {
1005 return Err(std::io::Error::new(error, message))
1009 if namespace == SCORER_PERSISTENCE_NAMESPACE && key == SCORER_PERSISTENCE_KEY {
1010 if let Some((error, message)) = self.scorer_error {
1011 return Err(std::io::Error::new(error, message))
1015 self.kv_store.write(namespace, key, buf)
1018 fn remove(&self, namespace: &str, key: &str) -> lightning::io::Result<()> {
1019 self.kv_store.remove(namespace, key)
1022 fn list(&self, namespace: &str) -> lightning::io::Result<Vec<String>> {
1023 self.kv_store.list(namespace)
1028 event_expectations: Option<VecDeque<TestResult>>,
1033 PaymentFailure { path: Path, short_channel_id: u64 },
1034 PaymentSuccess { path: Path },
1035 ProbeFailure { path: Path },
1036 ProbeSuccess { path: Path },
1041 Self { event_expectations: None }
1044 fn expect(&mut self, expectation: TestResult) {
1045 self.event_expectations.get_or_insert_with(VecDeque::new).push_back(expectation);
1049 impl lightning::util::ser::Writeable for TestScorer {
1050 fn write<W: lightning::util::ser::Writer>(&self, _: &mut W) -> Result<(), lightning::io::Error> { Ok(()) }
1053 impl Score for TestScorer {
1054 type ScoreParams = ();
1055 fn channel_penalty_msat(
1056 &self, _short_channel_id: u64, _source: &NodeId, _target: &NodeId, _usage: ChannelUsage, _score_params: &Self::ScoreParams
1057 ) -> u64 { unimplemented!(); }
1059 fn payment_path_failed(&mut self, actual_path: &Path, actual_short_channel_id: u64) {
1060 if let Some(expectations) = &mut self.event_expectations {
1061 match expectations.pop_front().unwrap() {
1062 TestResult::PaymentFailure { path, short_channel_id } => {
1063 assert_eq!(actual_path, &path);
1064 assert_eq!(actual_short_channel_id, short_channel_id);
1066 TestResult::PaymentSuccess { path } => {
1067 panic!("Unexpected successful payment path: {:?}", path)
1069 TestResult::ProbeFailure { path } => {
1070 panic!("Unexpected probe failure: {:?}", path)
1072 TestResult::ProbeSuccess { path } => {
1073 panic!("Unexpected probe success: {:?}", path)
1079 fn payment_path_successful(&mut self, actual_path: &Path) {
1080 if let Some(expectations) = &mut self.event_expectations {
1081 match expectations.pop_front().unwrap() {
1082 TestResult::PaymentFailure { path, .. } => {
1083 panic!("Unexpected payment path failure: {:?}", path)
1085 TestResult::PaymentSuccess { path } => {
1086 assert_eq!(actual_path, &path);
1088 TestResult::ProbeFailure { path } => {
1089 panic!("Unexpected probe failure: {:?}", path)
1091 TestResult::ProbeSuccess { path } => {
1092 panic!("Unexpected probe success: {:?}", path)
1098 fn probe_failed(&mut self, actual_path: &Path, _: u64) {
1099 if let Some(expectations) = &mut self.event_expectations {
1100 match expectations.pop_front().unwrap() {
1101 TestResult::PaymentFailure { path, .. } => {
1102 panic!("Unexpected payment path failure: {:?}", path)
1104 TestResult::PaymentSuccess { path } => {
1105 panic!("Unexpected payment path success: {:?}", path)
1107 TestResult::ProbeFailure { path } => {
1108 assert_eq!(actual_path, &path);
1110 TestResult::ProbeSuccess { path } => {
1111 panic!("Unexpected probe success: {:?}", path)
1116 fn probe_successful(&mut self, actual_path: &Path) {
1117 if let Some(expectations) = &mut self.event_expectations {
1118 match expectations.pop_front().unwrap() {
1119 TestResult::PaymentFailure { path, .. } => {
1120 panic!("Unexpected payment path failure: {:?}", path)
1122 TestResult::PaymentSuccess { path } => {
1123 panic!("Unexpected payment path success: {:?}", path)
1125 TestResult::ProbeFailure { path } => {
1126 panic!("Unexpected probe failure: {:?}", path)
1128 TestResult::ProbeSuccess { path } => {
1129 assert_eq!(actual_path, &path);
1136 impl Drop for TestScorer {
1137 fn drop(&mut self) {
1138 if std::thread::panicking() {
1142 if let Some(event_expectations) = &self.event_expectations {
1143 if !event_expectations.is_empty() {
1144 panic!("Unsatisfied event expectations: {:?}", event_expectations);
1150 fn get_full_filepath(filepath: String, filename: String) -> String {
1151 let mut path = PathBuf::from(filepath);
1152 path.push(filename);
1153 path.to_str().unwrap().to_string()
1156 fn create_nodes(num_nodes: usize, persist_dir: &str) -> (String, Vec<Node>) {
1157 let persist_temp_path = env::temp_dir().join(persist_dir);
1158 let persist_dir = persist_temp_path.to_string_lossy().to_string();
1159 let network = Network::Bitcoin;
1160 let mut nodes = Vec::new();
1161 for i in 0..num_nodes {
1162 let tx_broadcaster = Arc::new(test_utils::TestBroadcaster::new(network));
1163 let fee_estimator = Arc::new(test_utils::TestFeeEstimator { sat_per_kw: Mutex::new(253) });
1164 let logger = Arc::new(test_utils::TestLogger::with_id(format!("node {}", i)));
1165 let genesis_block = genesis_block(network);
1166 let network_graph = Arc::new(NetworkGraph::new(network, logger.clone()));
1167 let scorer = Arc::new(Mutex::new(TestScorer::new()));
1168 let seed = [i as u8; 32];
1169 let router = Arc::new(DefaultRouter::new(network_graph.clone(), logger.clone(), seed, scorer.clone(), ()));
1170 let chain_source = Arc::new(test_utils::TestChainSource::new(Network::Bitcoin));
1171 let kv_store = Arc::new(FilesystemStore::new(format!("{}_persister_{}", &persist_dir, i).into()));
1172 let now = Duration::from_secs(genesis_block.header.time as u64);
1173 let keys_manager = Arc::new(KeysManager::new(&seed, now.as_secs(), now.subsec_nanos()));
1174 let chain_monitor = Arc::new(chainmonitor::ChainMonitor::new(Some(chain_source.clone()), tx_broadcaster.clone(), logger.clone(), fee_estimator.clone(), kv_store.clone()));
1175 let best_block = BestBlock::from_network(network);
1176 let params = ChainParameters { network, best_block };
1177 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));
1178 let p2p_gossip_sync = Arc::new(P2PGossipSync::new(network_graph.clone(), Some(chain_source.clone()), logger.clone()));
1179 let rapid_gossip_sync = Arc::new(RapidGossipSync::new(network_graph.clone(), logger.clone()));
1180 let msg_handler = MessageHandler {
1181 chan_handler: Arc::new(test_utils::TestChannelMessageHandler::new(ChainHash::using_genesis_block(Network::Testnet))),
1182 route_handler: Arc::new(test_utils::TestRoutingMessageHandler::new()),
1183 onion_message_handler: IgnoringMessageHandler{}, custom_message_handler: IgnoringMessageHandler{}
1185 let peer_manager = Arc::new(PeerManager::new(msg_handler, 0, &seed, logger.clone(), keys_manager.clone()));
1186 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 };
1190 for i in 0..num_nodes {
1191 for j in (i+1)..num_nodes {
1192 nodes[i].node.peer_connected(&nodes[j].node.get_our_node_id(), &Init {
1193 features: nodes[j].node.init_features(), networks: None, remote_network_address: None
1195 nodes[j].node.peer_connected(&nodes[i].node.get_our_node_id(), &Init {
1196 features: nodes[i].node.init_features(), networks: None, remote_network_address: None
1201 (persist_dir, nodes)
1204 macro_rules! open_channel {
1205 ($node_a: expr, $node_b: expr, $channel_value: expr) => {{
1206 begin_open_channel!($node_a, $node_b, $channel_value);
1207 let events = $node_a.node.get_and_clear_pending_events();
1208 assert_eq!(events.len(), 1);
1209 let (temporary_channel_id, tx) = handle_funding_generation_ready!(events[0], $channel_value);
1210 $node_a.node.funding_transaction_generated(&temporary_channel_id, &$node_b.node.get_our_node_id(), tx.clone()).unwrap();
1211 $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()));
1212 get_event!($node_b, Event::ChannelPending);
1213 $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()));
1214 get_event!($node_a, Event::ChannelPending);
1219 macro_rules! begin_open_channel {
1220 ($node_a: expr, $node_b: expr, $channel_value: expr) => {{
1221 $node_a.node.create_channel($node_b.node.get_our_node_id(), $channel_value, 100, 42, None).unwrap();
1222 $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()));
1223 $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()));
1227 macro_rules! handle_funding_generation_ready {
1228 ($event: expr, $channel_value: expr) => {{
1230 Event::FundingGenerationReady { temporary_channel_id, channel_value_satoshis, ref output_script, user_channel_id, .. } => {
1231 assert_eq!(channel_value_satoshis, $channel_value);
1232 assert_eq!(user_channel_id, 42);
1234 let tx = Transaction { version: 1 as i32, lock_time: PackedLockTime(0), input: Vec::new(), output: vec![TxOut {
1235 value: channel_value_satoshis, script_pubkey: output_script.clone(),
1237 (temporary_channel_id, tx)
1239 _ => panic!("Unexpected event"),
1244 fn confirm_transaction_depth(node: &mut Node, tx: &Transaction, depth: u32) {
1245 for i in 1..=depth {
1246 let prev_blockhash = node.best_block.block_hash();
1247 let height = node.best_block.height() + 1;
1248 let header = create_dummy_header(prev_blockhash, height);
1249 let txdata = vec![(0, tx)];
1250 node.best_block = BestBlock::new(header.block_hash(), height);
1253 node.node.transactions_confirmed(&header, &txdata, height);
1254 node.chain_monitor.transactions_confirmed(&header, &txdata, height);
1256 x if x == depth => {
1257 node.node.best_block_updated(&header, height);
1258 node.chain_monitor.best_block_updated(&header, height);
1264 fn confirm_transaction(node: &mut Node, tx: &Transaction) {
1265 confirm_transaction_depth(node, tx, ANTI_REORG_DELAY);
1269 fn test_background_processor() {
1270 // Test that when a new channel is created, the ChannelManager needs to be re-persisted with
1271 // updates. Also test that when new updates are available, the manager signals that it needs
1272 // re-persistence and is successfully re-persisted.
1273 let (persist_dir, nodes) = create_nodes(2, "test_background_processor");
1275 // Go through the channel creation process so that each node has something to persist. Since
1276 // open_channel consumes events, it must complete before starting BackgroundProcessor to
1277 // avoid a race with processing events.
1278 let tx = open_channel!(nodes[0], nodes[1], 100000);
1280 // Initiate the background processors to watch each node.
1281 let data_dir = nodes[0].kv_store.get_data_dir();
1282 let persister = Arc::new(Persister::new(data_dir));
1283 let event_handler = |_: _| {};
1284 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()));
1286 macro_rules! check_persisted_data {
1287 ($node: expr, $filepath: expr) => {
1288 let mut expected_bytes = Vec::new();
1290 expected_bytes.clear();
1291 match $node.write(&mut expected_bytes) {
1293 match std::fs::read($filepath) {
1295 if bytes == expected_bytes {
1304 Err(e) => panic!("Unexpected error: {}", e)
1310 // Check that the initial channel manager data is persisted as expected.
1311 let filepath = get_full_filepath(format!("{}_persister_0", &persist_dir), "manager".to_string());
1312 check_persisted_data!(nodes[0].node, filepath.clone());
1315 if !nodes[0].node.get_persistence_condvar_value() { break }
1318 // Force-close the channel.
1319 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();
1321 // Check that the force-close updates are persisted.
1322 check_persisted_data!(nodes[0].node, filepath.clone());
1324 if !nodes[0].node.get_persistence_condvar_value() { break }
1327 // Check network graph is persisted
1328 let filepath = get_full_filepath(format!("{}_persister_0", &persist_dir), "network_graph".to_string());
1329 check_persisted_data!(nodes[0].network_graph, filepath.clone());
1331 // Check scorer is persisted
1332 let filepath = get_full_filepath(format!("{}_persister_0", &persist_dir), "scorer".to_string());
1333 check_persisted_data!(nodes[0].scorer, filepath.clone());
1335 if !std::thread::panicking() {
1336 bg_processor.stop().unwrap();
1341 fn test_timer_tick_called() {
1342 // Test that `ChannelManager::timer_tick_occurred` is called every `FRESHNESS_TIMER`,
1343 // `ChainMonitor::rebroadcast_pending_claims` is called every `REBROADCAST_TIMER`, and
1344 // `PeerManager::timer_tick_occurred` every `PING_TIMER`.
1345 let (_, nodes) = create_nodes(1, "test_timer_tick_called");
1346 let data_dir = nodes[0].kv_store.get_data_dir();
1347 let persister = Arc::new(Persister::new(data_dir));
1348 let event_handler = |_: _| {};
1349 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()));
1351 let log_entries = nodes[0].logger.lines.lock().unwrap();
1352 let desired_log_1 = "Calling ChannelManager's timer_tick_occurred".to_string();
1353 let desired_log_2 = "Calling PeerManager's timer_tick_occurred".to_string();
1354 let desired_log_3 = "Rebroadcasting monitor's pending claims".to_string();
1355 if log_entries.get(&("lightning_background_processor".to_string(), desired_log_1)).is_some() &&
1356 log_entries.get(&("lightning_background_processor".to_string(), desired_log_2)).is_some() &&
1357 log_entries.get(&("lightning_background_processor".to_string(), desired_log_3)).is_some() {
1362 if !std::thread::panicking() {
1363 bg_processor.stop().unwrap();
1368 fn test_channel_manager_persist_error() {
1369 // Test that if we encounter an error during manager persistence, the thread panics.
1370 let (_, nodes) = create_nodes(2, "test_persist_error");
1371 open_channel!(nodes[0], nodes[1], 100000);
1373 let data_dir = nodes[0].kv_store.get_data_dir();
1374 let persister = Arc::new(Persister::new(data_dir).with_manager_error(std::io::ErrorKind::Other, "test"));
1375 let event_handler = |_: _| {};
1376 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()));
1377 match bg_processor.join() {
1378 Ok(_) => panic!("Expected error persisting manager"),
1380 assert_eq!(e.kind(), std::io::ErrorKind::Other);
1381 assert_eq!(e.get_ref().unwrap().to_string(), "test");
1387 #[cfg(feature = "futures")]
1388 async fn test_channel_manager_persist_error_async() {
1389 // Test that if we encounter an error during manager persistence, the thread panics.
1390 let (_, nodes) = create_nodes(2, "test_persist_error_sync");
1391 open_channel!(nodes[0], nodes[1], 100000);
1393 let data_dir = nodes[0].kv_store.get_data_dir();
1394 let persister = Arc::new(Persister::new(data_dir).with_manager_error(std::io::ErrorKind::Other, "test"));
1396 let bp_future = super::process_events_async(
1397 persister, |_: _| {async {}}, nodes[0].chain_monitor.clone(), nodes[0].node.clone(),
1398 nodes[0].rapid_gossip_sync(), nodes[0].peer_manager.clone(), nodes[0].logger.clone(),
1399 Some(nodes[0].scorer.clone()), move |dur: Duration| {
1400 Box::pin(async move {
1401 tokio::time::sleep(dur).await;
1406 match bp_future.await {
1407 Ok(_) => panic!("Expected error persisting manager"),
1409 assert_eq!(e.kind(), std::io::ErrorKind::Other);
1410 assert_eq!(e.get_ref().unwrap().to_string(), "test");
1416 fn test_network_graph_persist_error() {
1417 // Test that if we encounter an error during network graph persistence, an error gets returned.
1418 let (_, nodes) = create_nodes(2, "test_persist_network_graph_error");
1419 let data_dir = nodes[0].kv_store.get_data_dir();
1420 let persister = Arc::new(Persister::new(data_dir).with_graph_error(std::io::ErrorKind::Other, "test"));
1421 let event_handler = |_: _| {};
1422 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()));
1424 match bg_processor.stop() {
1425 Ok(_) => panic!("Expected error persisting network graph"),
1427 assert_eq!(e.kind(), std::io::ErrorKind::Other);
1428 assert_eq!(e.get_ref().unwrap().to_string(), "test");
1434 fn test_scorer_persist_error() {
1435 // Test that if we encounter an error during scorer persistence, an error gets returned.
1436 let (_, nodes) = create_nodes(2, "test_persist_scorer_error");
1437 let data_dir = nodes[0].kv_store.get_data_dir();
1438 let persister = Arc::new(Persister::new(data_dir).with_scorer_error(std::io::ErrorKind::Other, "test"));
1439 let event_handler = |_: _| {};
1440 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()));
1442 match bg_processor.stop() {
1443 Ok(_) => panic!("Expected error persisting scorer"),
1445 assert_eq!(e.kind(), std::io::ErrorKind::Other);
1446 assert_eq!(e.get_ref().unwrap().to_string(), "test");
1452 fn test_background_event_handling() {
1453 let (_, mut nodes) = create_nodes(2, "test_background_event_handling");
1454 let channel_value = 100000;
1455 let data_dir = nodes[0].kv_store.get_data_dir();
1456 let persister = Arc::new(Persister::new(data_dir.clone()));
1458 // Set up a background event handler for FundingGenerationReady events.
1459 let (funding_generation_send, funding_generation_recv) = std::sync::mpsc::sync_channel(1);
1460 let (channel_pending_send, channel_pending_recv) = std::sync::mpsc::sync_channel(1);
1461 let event_handler = move |event: Event| match event {
1462 Event::FundingGenerationReady { .. } => funding_generation_send.send(handle_funding_generation_ready!(event, channel_value)).unwrap(),
1463 Event::ChannelPending { .. } => channel_pending_send.send(()).unwrap(),
1464 Event::ChannelReady { .. } => {},
1465 _ => panic!("Unexpected event: {:?}", event),
1468 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()));
1470 // Open a channel and check that the FundingGenerationReady event was handled.
1471 begin_open_channel!(nodes[0], nodes[1], channel_value);
1472 let (temporary_channel_id, funding_tx) = funding_generation_recv
1473 .recv_timeout(Duration::from_secs(EVENT_DEADLINE))
1474 .expect("FundingGenerationReady not handled within deadline");
1475 nodes[0].node.funding_transaction_generated(&temporary_channel_id, &nodes[1].node.get_our_node_id(), funding_tx.clone()).unwrap();
1476 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()));
1477 get_event!(nodes[1], Event::ChannelPending);
1478 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()));
1479 let _ = channel_pending_recv.recv_timeout(Duration::from_secs(EVENT_DEADLINE))
1480 .expect("ChannelPending not handled within deadline");
1482 // Confirm the funding transaction.
1483 confirm_transaction(&mut nodes[0], &funding_tx);
1484 let as_funding = get_event_msg!(nodes[0], MessageSendEvent::SendChannelReady, nodes[1].node.get_our_node_id());
1485 confirm_transaction(&mut nodes[1], &funding_tx);
1486 let bs_funding = get_event_msg!(nodes[1], MessageSendEvent::SendChannelReady, nodes[0].node.get_our_node_id());
1487 nodes[0].node.handle_channel_ready(&nodes[1].node.get_our_node_id(), &bs_funding);
1488 let _as_channel_update = get_event_msg!(nodes[0], MessageSendEvent::SendChannelUpdate, nodes[1].node.get_our_node_id());
1489 nodes[1].node.handle_channel_ready(&nodes[0].node.get_our_node_id(), &as_funding);
1490 let _bs_channel_update = get_event_msg!(nodes[1], MessageSendEvent::SendChannelUpdate, nodes[0].node.get_our_node_id());
1492 if !std::thread::panicking() {
1493 bg_processor.stop().unwrap();
1496 // Set up a background event handler for SpendableOutputs events.
1497 let (sender, receiver) = std::sync::mpsc::sync_channel(1);
1498 let event_handler = move |event: Event| match event {
1499 Event::SpendableOutputs { .. } => sender.send(event).unwrap(),
1500 Event::ChannelReady { .. } => {},
1501 Event::ChannelClosed { .. } => {},
1502 _ => panic!("Unexpected event: {:?}", event),
1504 let persister = Arc::new(Persister::new(data_dir));
1505 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()));
1507 // Force close the channel and check that the SpendableOutputs event was handled.
1508 nodes[0].node.force_close_broadcasting_latest_txn(&nodes[0].node.list_channels()[0].channel_id, &nodes[1].node.get_our_node_id()).unwrap();
1509 let commitment_tx = nodes[0].tx_broadcaster.txn_broadcasted.lock().unwrap().pop().unwrap();
1510 confirm_transaction_depth(&mut nodes[0], &commitment_tx, BREAKDOWN_TIMEOUT as u32);
1512 let event = receiver
1513 .recv_timeout(Duration::from_secs(EVENT_DEADLINE))
1514 .expect("Events not handled within deadline");
1516 Event::SpendableOutputs { .. } => {},
1517 _ => panic!("Unexpected event: {:?}", event),
1520 if !std::thread::panicking() {
1521 bg_processor.stop().unwrap();
1526 fn test_scorer_persistence() {
1527 let (_, nodes) = create_nodes(2, "test_scorer_persistence");
1528 let data_dir = nodes[0].kv_store.get_data_dir();
1529 let persister = Arc::new(Persister::new(data_dir));
1530 let event_handler = |_: _| {};
1531 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()));
1534 let log_entries = nodes[0].logger.lines.lock().unwrap();
1535 let expected_log = "Persisting scorer".to_string();
1536 if log_entries.get(&("lightning_background_processor".to_string(), expected_log)).is_some() {
1541 if !std::thread::panicking() {
1542 bg_processor.stop().unwrap();
1546 macro_rules! do_test_not_pruning_network_graph_until_graph_sync_completion {
1547 ($nodes: expr, $receive: expr, $sleep: expr) => {
1548 let features = ChannelFeatures::empty();
1549 $nodes[0].network_graph.add_channel_from_partial_announcement(
1550 42, 53, features, $nodes[0].node.get_our_node_id(), $nodes[1].node.get_our_node_id()
1551 ).expect("Failed to update channel from partial announcement");
1552 let original_graph_description = $nodes[0].network_graph.to_string();
1553 assert!(original_graph_description.contains("42: features: 0000, node_one:"));
1554 assert_eq!($nodes[0].network_graph.read_only().channels().len(), 1);
1558 let log_entries = $nodes[0].logger.lines.lock().unwrap();
1559 let loop_counter = "Calling ChannelManager's timer_tick_occurred".to_string();
1560 if *log_entries.get(&("lightning_background_processor".to_string(), loop_counter))
1563 // Wait until the loop has gone around at least twice.
1568 let initialization_input = vec![
1569 76, 68, 75, 1, 111, 226, 140, 10, 182, 241, 179, 114, 193, 166, 162, 70, 174, 99, 247,
1570 79, 147, 30, 131, 101, 225, 90, 8, 156, 104, 214, 25, 0, 0, 0, 0, 0, 97, 227, 98, 218,
1571 0, 0, 0, 4, 2, 22, 7, 207, 206, 25, 164, 197, 231, 230, 231, 56, 102, 61, 250, 251,
1572 187, 172, 38, 46, 79, 247, 108, 44, 155, 48, 219, 238, 252, 53, 192, 6, 67, 2, 36, 125,
1573 157, 176, 223, 175, 234, 116, 94, 248, 201, 225, 97, 235, 50, 47, 115, 172, 63, 136,
1574 88, 216, 115, 11, 111, 217, 114, 84, 116, 124, 231, 107, 2, 158, 1, 242, 121, 152, 106,
1575 204, 131, 186, 35, 93, 70, 216, 10, 237, 224, 183, 89, 95, 65, 3, 83, 185, 58, 138,
1576 181, 64, 187, 103, 127, 68, 50, 2, 201, 19, 17, 138, 136, 149, 185, 226, 156, 137, 175,
1577 110, 32, 237, 0, 217, 90, 31, 100, 228, 149, 46, 219, 175, 168, 77, 4, 143, 38, 128,
1578 76, 97, 0, 0, 0, 2, 0, 0, 255, 8, 153, 192, 0, 2, 27, 0, 0, 0, 1, 0, 0, 255, 2, 68,
1579 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,
1580 0, 0, 0, 1, 0, 0, 0, 0, 58, 85, 116, 216, 255, 8, 153, 192, 0, 2, 27, 0, 0, 25, 0, 0,
1581 0, 1, 0, 0, 0, 125, 255, 2, 68, 226, 0, 6, 11, 0, 1, 5, 0, 0, 0, 0, 29, 129, 25, 192,
1583 $nodes[0].rapid_gossip_sync.update_network_graph_no_std(&initialization_input[..], Some(1642291930)).unwrap();
1585 // this should have added two channels and pruned the previous one.
1586 assert_eq!($nodes[0].network_graph.read_only().channels().len(), 2);
1588 $receive.expect("Network graph not pruned within deadline");
1590 // all channels should now be pruned
1591 assert_eq!($nodes[0].network_graph.read_only().channels().len(), 0);
1596 fn test_not_pruning_network_graph_until_graph_sync_completion() {
1597 let (sender, receiver) = std::sync::mpsc::sync_channel(1);
1599 let (_, nodes) = create_nodes(2, "test_not_pruning_network_graph_until_graph_sync_completion");
1600 let data_dir = nodes[0].kv_store.get_data_dir();
1601 let persister = Arc::new(Persister::new(data_dir).with_graph_persistence_notifier(sender));
1603 let event_handler = |_: _| {};
1604 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()));
1606 do_test_not_pruning_network_graph_until_graph_sync_completion!(nodes,
1607 receiver.recv_timeout(Duration::from_secs(super::FIRST_NETWORK_PRUNE_TIMER * 5)),
1608 std::thread::sleep(Duration::from_millis(1)));
1610 background_processor.stop().unwrap();
1614 #[cfg(feature = "futures")]
1615 async fn test_not_pruning_network_graph_until_graph_sync_completion_async() {
1616 let (sender, receiver) = std::sync::mpsc::sync_channel(1);
1618 let (_, nodes) = create_nodes(2, "test_not_pruning_network_graph_until_graph_sync_completion_async");
1619 let data_dir = nodes[0].kv_store.get_data_dir();
1620 let persister = Arc::new(Persister::new(data_dir).with_graph_persistence_notifier(sender));
1622 let (exit_sender, exit_receiver) = tokio::sync::watch::channel(());
1623 let bp_future = super::process_events_async(
1624 persister, |_: _| {async {}}, nodes[0].chain_monitor.clone(), nodes[0].node.clone(),
1625 nodes[0].rapid_gossip_sync(), nodes[0].peer_manager.clone(), nodes[0].logger.clone(),
1626 Some(nodes[0].scorer.clone()), move |dur: Duration| {
1627 let mut exit_receiver = exit_receiver.clone();
1628 Box::pin(async move {
1630 _ = tokio::time::sleep(dur) => false,
1631 _ = exit_receiver.changed() => true,
1637 let t1 = tokio::spawn(bp_future);
1638 let t2 = tokio::spawn(async move {
1639 do_test_not_pruning_network_graph_until_graph_sync_completion!(nodes, {
1642 tokio::time::sleep(Duration::from_secs(super::FIRST_NETWORK_PRUNE_TIMER)).await;
1643 if let Ok(()) = receiver.try_recv() { break Ok::<(), ()>(()); }
1647 }, tokio::time::sleep(Duration::from_millis(1)).await);
1648 exit_sender.send(()).unwrap();
1650 let (r1, r2) = tokio::join!(t1, t2);
1651 r1.unwrap().unwrap();
1655 macro_rules! do_test_payment_path_scoring {
1656 ($nodes: expr, $receive: expr) => {
1657 // Ensure that we update the scorer when relevant events are processed. In this case, we ensure
1658 // that we update the scorer upon a payment path succeeding (note that the channel must be
1659 // public or else we won't score it).
1660 // A background event handler for FundingGenerationReady events must be hooked up to a
1661 // running background processor.
1662 let scored_scid = 4242;
1663 let secp_ctx = Secp256k1::new();
1664 let node_1_privkey = SecretKey::from_slice(&[42; 32]).unwrap();
1665 let node_1_id = PublicKey::from_secret_key(&secp_ctx, &node_1_privkey);
1667 let path = Path { hops: vec![RouteHop {
1669 node_features: NodeFeatures::empty(),
1670 short_channel_id: scored_scid,
1671 channel_features: ChannelFeatures::empty(),
1673 cltv_expiry_delta: MIN_CLTV_EXPIRY_DELTA as u32,
1674 }], blinded_tail: None };
1676 $nodes[0].scorer.lock().unwrap().expect(TestResult::PaymentFailure { path: path.clone(), short_channel_id: scored_scid });
1677 $nodes[0].node.push_pending_event(Event::PaymentPathFailed {
1679 payment_hash: PaymentHash([42; 32]),
1680 payment_failed_permanently: false,
1681 failure: PathFailure::OnPath { network_update: None },
1683 short_channel_id: Some(scored_scid),
1685 let event = $receive.expect("PaymentPathFailed not handled within deadline");
1687 Event::PaymentPathFailed { .. } => {},
1688 _ => panic!("Unexpected event"),
1691 // Ensure we'll score payments that were explicitly failed back by the destination as
1693 $nodes[0].scorer.lock().unwrap().expect(TestResult::ProbeSuccess { path: path.clone() });
1694 $nodes[0].node.push_pending_event(Event::PaymentPathFailed {
1696 payment_hash: PaymentHash([42; 32]),
1697 payment_failed_permanently: true,
1698 failure: PathFailure::OnPath { network_update: None },
1700 short_channel_id: None,
1702 let event = $receive.expect("PaymentPathFailed not handled within deadline");
1704 Event::PaymentPathFailed { .. } => {},
1705 _ => panic!("Unexpected event"),
1708 $nodes[0].scorer.lock().unwrap().expect(TestResult::PaymentSuccess { path: path.clone() });
1709 $nodes[0].node.push_pending_event(Event::PaymentPathSuccessful {
1710 payment_id: PaymentId([42; 32]),
1714 let event = $receive.expect("PaymentPathSuccessful not handled within deadline");
1716 Event::PaymentPathSuccessful { .. } => {},
1717 _ => panic!("Unexpected event"),
1720 $nodes[0].scorer.lock().unwrap().expect(TestResult::ProbeSuccess { path: path.clone() });
1721 $nodes[0].node.push_pending_event(Event::ProbeSuccessful {
1722 payment_id: PaymentId([42; 32]),
1723 payment_hash: PaymentHash([42; 32]),
1726 let event = $receive.expect("ProbeSuccessful not handled within deadline");
1728 Event::ProbeSuccessful { .. } => {},
1729 _ => panic!("Unexpected event"),
1732 $nodes[0].scorer.lock().unwrap().expect(TestResult::ProbeFailure { path: path.clone() });
1733 $nodes[0].node.push_pending_event(Event::ProbeFailed {
1734 payment_id: PaymentId([42; 32]),
1735 payment_hash: PaymentHash([42; 32]),
1737 short_channel_id: Some(scored_scid),
1739 let event = $receive.expect("ProbeFailure not handled within deadline");
1741 Event::ProbeFailed { .. } => {},
1742 _ => panic!("Unexpected event"),
1748 fn test_payment_path_scoring() {
1749 let (sender, receiver) = std::sync::mpsc::sync_channel(1);
1750 let event_handler = move |event: Event| match event {
1751 Event::PaymentPathFailed { .. } => sender.send(event).unwrap(),
1752 Event::PaymentPathSuccessful { .. } => sender.send(event).unwrap(),
1753 Event::ProbeSuccessful { .. } => sender.send(event).unwrap(),
1754 Event::ProbeFailed { .. } => sender.send(event).unwrap(),
1755 _ => panic!("Unexpected event: {:?}", event),
1758 let (_, nodes) = create_nodes(1, "test_payment_path_scoring");
1759 let data_dir = nodes[0].kv_store.get_data_dir();
1760 let persister = Arc::new(Persister::new(data_dir));
1761 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()));
1763 do_test_payment_path_scoring!(nodes, receiver.recv_timeout(Duration::from_secs(EVENT_DEADLINE)));
1765 if !std::thread::panicking() {
1766 bg_processor.stop().unwrap();
1769 let log_entries = nodes[0].logger.lines.lock().unwrap();
1770 let expected_log = "Persisting scorer after update".to_string();
1771 assert_eq!(*log_entries.get(&("lightning_background_processor".to_string(), expected_log)).unwrap(), 5);
1775 #[cfg(feature = "futures")]
1776 async fn test_payment_path_scoring_async() {
1777 let (sender, mut receiver) = tokio::sync::mpsc::channel(1);
1778 let event_handler = move |event: Event| {
1779 let sender_ref = sender.clone();
1782 Event::PaymentPathFailed { .. } => { sender_ref.send(event).await.unwrap() },
1783 Event::PaymentPathSuccessful { .. } => { sender_ref.send(event).await.unwrap() },
1784 Event::ProbeSuccessful { .. } => { sender_ref.send(event).await.unwrap() },
1785 Event::ProbeFailed { .. } => { sender_ref.send(event).await.unwrap() },
1786 _ => panic!("Unexpected event: {:?}", event),
1791 let (_, nodes) = create_nodes(1, "test_payment_path_scoring_async");
1792 let data_dir = nodes[0].kv_store.get_data_dir();
1793 let persister = Arc::new(Persister::new(data_dir));
1795 let (exit_sender, exit_receiver) = tokio::sync::watch::channel(());
1797 let bp_future = super::process_events_async(
1798 persister, event_handler, nodes[0].chain_monitor.clone(), nodes[0].node.clone(),
1799 nodes[0].no_gossip_sync(), nodes[0].peer_manager.clone(), nodes[0].logger.clone(),
1800 Some(nodes[0].scorer.clone()), move |dur: Duration| {
1801 let mut exit_receiver = exit_receiver.clone();
1802 Box::pin(async move {
1804 _ = tokio::time::sleep(dur) => false,
1805 _ = exit_receiver.changed() => true,
1810 let t1 = tokio::spawn(bp_future);
1811 let t2 = tokio::spawn(async move {
1812 do_test_payment_path_scoring!(nodes, receiver.recv().await);
1813 exit_sender.send(()).unwrap();
1815 let log_entries = nodes[0].logger.lines.lock().unwrap();
1816 let expected_log = "Persisting scorer after update".to_string();
1817 assert_eq!(*log_entries.get(&("lightning_background_processor".to_string(), expected_log)).unwrap(), 5);
1820 let (r1, r2) = tokio::join!(t1, t2);
1821 r1.unwrap().unwrap();