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 #![deny(rustdoc::broken_intra_doc_links)]
6 #![deny(rustdoc::private_intra_doc_links)]
9 #![cfg_attr(not(feature = "futures"), deny(unsafe_code))]
11 #![cfg_attr(docsrs, feature(doc_auto_cfg))]
13 #![cfg_attr(all(not(feature = "std"), not(test)), no_std)]
15 #[cfg(any(test, feature = "std"))]
18 #[cfg(not(feature = "std"))]
21 #[macro_use] extern crate lightning;
22 extern crate lightning_rapid_gossip_sync;
25 use lightning::chain::chaininterface::{BroadcasterInterface, FeeEstimator};
26 use lightning::chain::chainmonitor::{ChainMonitor, Persist};
27 use lightning::events::{Event, PathFailure};
28 #[cfg(feature = "std")]
29 use lightning::events::EventHandler;
30 #[cfg(any(feature = "std", feature = "futures"))]
31 use lightning::events::EventsProvider;
33 use lightning::ln::channelmanager::AChannelManager;
34 use lightning::ln::msgs::OnionMessageHandler;
35 use lightning::ln::peer_handler::APeerManager;
36 use lightning::routing::gossip::{NetworkGraph, P2PGossipSync};
37 use lightning::routing::utxo::UtxoLookup;
38 use lightning::routing::scoring::{ScoreUpdate, WriteableScore};
39 use lightning::util::logger::Logger;
40 use lightning::util::persist::Persister;
41 #[cfg(feature = "std")]
42 use lightning::util::wakers::Sleeper;
43 use lightning_rapid_gossip_sync::RapidGossipSync;
46 use core::time::Duration;
48 #[cfg(feature = "std")]
50 #[cfg(feature = "std")]
51 use core::sync::atomic::{AtomicBool, Ordering};
52 #[cfg(feature = "std")]
53 use std::thread::{self, JoinHandle};
54 #[cfg(feature = "std")]
55 use std::time::Instant;
57 #[cfg(not(feature = "std"))]
60 /// `BackgroundProcessor` takes care of tasks that (1) need to happen periodically to keep
61 /// Rust-Lightning running properly, and (2) either can or should be run in the background. Its
62 /// responsibilities are:
63 /// * Processing [`Event`]s with a user-provided [`EventHandler`].
64 /// * Monitoring whether the [`ChannelManager`] needs to be re-persisted to disk, and if so,
65 /// writing it to disk/backups by invoking the callback given to it at startup.
66 /// [`ChannelManager`] persistence should be done in the background.
67 /// * Calling [`ChannelManager::timer_tick_occurred`], [`ChainMonitor::rebroadcast_pending_claims`]
68 /// and [`PeerManager::timer_tick_occurred`] at the appropriate intervals.
69 /// * Calling [`NetworkGraph::remove_stale_channels_and_tracking`] (if a [`GossipSync`] with a
70 /// [`NetworkGraph`] is provided to [`BackgroundProcessor::start`]).
72 /// It will also call [`PeerManager::process_events`] periodically though this shouldn't be relied
73 /// upon as doing so may result in high latency.
77 /// If [`ChannelManager`] persistence fails and the persisted manager becomes out-of-date, then
78 /// there is a risk of channels force-closing on startup when the manager realizes it's outdated.
79 /// However, as long as [`ChannelMonitor`] backups are sound, no funds besides those used for
80 /// unilateral chain closure fees are at risk.
82 /// [`ChannelManager`]: lightning::ln::channelmanager::ChannelManager
83 /// [`ChannelManager::timer_tick_occurred`]: lightning::ln::channelmanager::ChannelManager::timer_tick_occurred
84 /// [`ChannelMonitor`]: lightning::chain::channelmonitor::ChannelMonitor
85 /// [`Event`]: lightning::events::Event
86 /// [`PeerManager::timer_tick_occurred`]: lightning::ln::peer_handler::PeerManager::timer_tick_occurred
87 /// [`PeerManager::process_events`]: lightning::ln::peer_handler::PeerManager::process_events
88 #[cfg(feature = "std")]
89 #[must_use = "BackgroundProcessor will immediately stop on drop. It should be stored until shutdown."]
90 pub struct BackgroundProcessor {
91 stop_thread: Arc<AtomicBool>,
92 thread_handle: Option<JoinHandle<Result<(), std::io::Error>>>,
96 const FRESHNESS_TIMER: u64 = 60;
98 const FRESHNESS_TIMER: u64 = 1;
100 #[cfg(all(not(test), not(debug_assertions)))]
101 const PING_TIMER: u64 = 10;
102 /// Signature operations take a lot longer without compiler optimisations.
103 /// Increasing the ping timer allows for this but slower devices will be disconnected if the
104 /// timeout is reached.
105 #[cfg(all(not(test), debug_assertions))]
106 const PING_TIMER: u64 = 30;
108 const PING_TIMER: u64 = 1;
111 const ONION_MESSAGE_HANDLER_TIMER: u64 = 10;
113 const ONION_MESSAGE_HANDLER_TIMER: u64 = 1;
115 /// Prune the network graph of stale entries hourly.
116 const NETWORK_PRUNE_TIMER: u64 = 60 * 60;
119 const SCORER_PERSIST_TIMER: u64 = 60 * 5;
121 const SCORER_PERSIST_TIMER: u64 = 1;
124 const FIRST_NETWORK_PRUNE_TIMER: u64 = 60;
126 const FIRST_NETWORK_PRUNE_TIMER: u64 = 1;
129 const REBROADCAST_TIMER: u64 = 30;
131 const REBROADCAST_TIMER: u64 = 1;
133 #[cfg(feature = "futures")]
134 /// core::cmp::min is not currently const, so we define a trivial (and equivalent) replacement
135 const fn min_u64(a: u64, b: u64) -> u64 { if a < b { a } else { b } }
136 #[cfg(feature = "futures")]
137 const FASTEST_TIMER: u64 = min_u64(min_u64(FRESHNESS_TIMER, PING_TIMER),
138 min_u64(SCORER_PERSIST_TIMER, min_u64(FIRST_NETWORK_PRUNE_TIMER, REBROADCAST_TIMER)));
140 /// Either [`P2PGossipSync`] or [`RapidGossipSync`].
142 P: Deref<Target = P2PGossipSync<G, U, L>>,
143 R: Deref<Target = RapidGossipSync<G, L>>,
144 G: Deref<Target = NetworkGraph<L>>,
148 where U::Target: UtxoLookup, L::Target: Logger {
149 /// Gossip sync via the lightning peer-to-peer network as defined by BOLT 7.
151 /// Rapid gossip sync from a trusted server.
158 P: Deref<Target = P2PGossipSync<G, U, L>>,
159 R: Deref<Target = RapidGossipSync<G, L>>,
160 G: Deref<Target = NetworkGraph<L>>,
163 > GossipSync<P, R, G, U, L>
164 where U::Target: UtxoLookup, L::Target: Logger {
165 fn network_graph(&self) -> Option<&G> {
167 GossipSync::P2P(gossip_sync) => Some(gossip_sync.network_graph()),
168 GossipSync::Rapid(gossip_sync) => Some(gossip_sync.network_graph()),
169 GossipSync::None => None,
173 fn prunable_network_graph(&self) -> Option<&G> {
175 GossipSync::P2P(gossip_sync) => Some(gossip_sync.network_graph()),
176 GossipSync::Rapid(gossip_sync) => {
177 if gossip_sync.is_initial_sync_complete() {
178 Some(gossip_sync.network_graph())
183 GossipSync::None => None,
188 /// This is not exported to bindings users as the bindings concretize everything and have constructors for us
189 impl<P: Deref<Target = P2PGossipSync<G, U, L>>, G: Deref<Target = NetworkGraph<L>>, U: Deref, L: Deref>
190 GossipSync<P, &RapidGossipSync<G, L>, G, U, L>
192 U::Target: UtxoLookup,
195 /// Initializes a new [`GossipSync::P2P`] variant.
196 pub fn p2p(gossip_sync: P) -> Self {
197 GossipSync::P2P(gossip_sync)
201 /// This is not exported to bindings users as the bindings concretize everything and have constructors for us
202 impl<'a, R: Deref<Target = RapidGossipSync<G, L>>, G: Deref<Target = NetworkGraph<L>>, L: Deref>
204 &P2PGossipSync<G, &'a (dyn UtxoLookup + Send + Sync), L>,
207 &'a (dyn UtxoLookup + Send + Sync),
213 /// Initializes a new [`GossipSync::Rapid`] variant.
214 pub fn rapid(gossip_sync: R) -> Self {
215 GossipSync::Rapid(gossip_sync)
219 /// This is not exported to bindings users as the bindings concretize everything and have constructors for us
222 &P2PGossipSync<&'a NetworkGraph<L>, &'a (dyn UtxoLookup + Send + Sync), L>,
223 &RapidGossipSync<&'a NetworkGraph<L>, L>,
225 &'a (dyn UtxoLookup + Send + Sync),
231 /// Initializes a new [`GossipSync::None`] variant.
232 pub fn none() -> Self {
237 fn handle_network_graph_update<L: Deref>(
238 network_graph: &NetworkGraph<L>, event: &Event
239 ) where L::Target: Logger {
240 if let Event::PaymentPathFailed {
241 failure: PathFailure::OnPath { network_update: Some(ref upd) }, .. } = event
243 network_graph.handle_network_update(upd);
247 /// Updates scorer based on event and returns whether an update occurred so we can decide whether
249 fn update_scorer<'a, S: 'static + Deref<Target = SC> + Send + Sync, SC: 'a + WriteableScore<'a>>(
250 scorer: &'a S, event: &Event, duration_since_epoch: Duration,
253 Event::PaymentPathFailed { ref path, short_channel_id: Some(scid), .. } => {
254 let mut score = scorer.write_lock();
255 score.payment_path_failed(path, *scid, duration_since_epoch);
257 Event::PaymentPathFailed { ref path, payment_failed_permanently: true, .. } => {
258 // Reached if the destination explicitly failed it back. We treat this as a successful probe
259 // because the payment made it all the way to the destination with sufficient liquidity.
260 let mut score = scorer.write_lock();
261 score.probe_successful(path, duration_since_epoch);
263 Event::PaymentPathSuccessful { path, .. } => {
264 let mut score = scorer.write_lock();
265 score.payment_path_successful(path, duration_since_epoch);
267 Event::ProbeSuccessful { path, .. } => {
268 let mut score = scorer.write_lock();
269 score.probe_successful(path, duration_since_epoch);
271 Event::ProbeFailed { path, short_channel_id: Some(scid), .. } => {
272 let mut score = scorer.write_lock();
273 score.probe_failed(path, *scid, duration_since_epoch);
280 macro_rules! define_run_body {
282 $persister: ident, $chain_monitor: ident, $process_chain_monitor_events: expr,
283 $channel_manager: ident, $process_channel_manager_events: expr,
284 $peer_manager: ident, $process_onion_message_handler_events: expr, $gossip_sync: ident,
285 $logger: ident, $scorer: ident, $loop_exit_check: expr, $await: expr, $get_timer: expr,
286 $timer_elapsed: expr, $check_slow_await: expr, $time_fetch: expr,
288 log_trace!($logger, "Calling ChannelManager's timer_tick_occurred on startup");
289 $channel_manager.get_cm().timer_tick_occurred();
290 log_trace!($logger, "Rebroadcasting monitor's pending claims on startup");
291 $chain_monitor.rebroadcast_pending_claims();
293 let mut last_freshness_call = $get_timer(FRESHNESS_TIMER);
294 let mut last_onion_message_handler_call = $get_timer(ONION_MESSAGE_HANDLER_TIMER);
295 let mut last_ping_call = $get_timer(PING_TIMER);
296 let mut last_prune_call = $get_timer(FIRST_NETWORK_PRUNE_TIMER);
297 let mut last_scorer_persist_call = $get_timer(SCORER_PERSIST_TIMER);
298 let mut last_rebroadcast_call = $get_timer(REBROADCAST_TIMER);
299 let mut have_pruned = false;
300 let mut have_decayed_scorer = false;
303 $process_channel_manager_events;
304 $process_chain_monitor_events;
305 $process_onion_message_handler_events;
307 // Note that the PeerManager::process_events may block on ChannelManager's locks,
308 // hence it comes last here. When the ChannelManager finishes whatever it's doing,
309 // we want to ensure we get into `persist_manager` as quickly as we can, especially
310 // without running the normal event processing above and handing events to users.
312 // Specifically, on an *extremely* slow machine, we may see ChannelManager start
313 // processing a message effectively at any point during this loop. In order to
314 // minimize the time between such processing completing and persisting the updated
315 // ChannelManager, we want to minimize methods blocking on a ChannelManager
316 // generally, and as a fallback place such blocking only immediately before
318 $peer_manager.as_ref().process_events();
320 // Exit the loop if the background processor was requested to stop.
321 if $loop_exit_check {
322 log_trace!($logger, "Terminating background processor.");
326 // We wait up to 100ms, but track how long it takes to detect being put to sleep,
327 // see `await_start`'s use below.
328 let mut await_start = None;
329 if $check_slow_await { await_start = Some($get_timer(1)); }
331 let await_slow = if $check_slow_await { $timer_elapsed(&mut await_start.unwrap(), 1) } else { false };
333 // Exit the loop if the background processor was requested to stop.
334 if $loop_exit_check {
335 log_trace!($logger, "Terminating background processor.");
339 if $channel_manager.get_cm().get_and_clear_needs_persistence() {
340 log_trace!($logger, "Persisting ChannelManager...");
341 $persister.persist_manager(&$channel_manager)?;
342 log_trace!($logger, "Done persisting ChannelManager.");
344 if $timer_elapsed(&mut last_freshness_call, FRESHNESS_TIMER) {
345 log_trace!($logger, "Calling ChannelManager's timer_tick_occurred");
346 $channel_manager.get_cm().timer_tick_occurred();
347 last_freshness_call = $get_timer(FRESHNESS_TIMER);
349 if $timer_elapsed(&mut last_onion_message_handler_call, ONION_MESSAGE_HANDLER_TIMER) {
350 log_trace!($logger, "Calling OnionMessageHandler's timer_tick_occurred");
351 $peer_manager.onion_message_handler().timer_tick_occurred();
352 last_onion_message_handler_call = $get_timer(ONION_MESSAGE_HANDLER_TIMER);
355 // On various platforms, we may be starved of CPU cycles for several reasons.
356 // E.g. on iOS, if we've been in the background, we will be entirely paused.
357 // Similarly, if we're on a desktop platform and the device has been asleep, we
358 // may not get any cycles.
359 // We detect this by checking if our max-100ms-sleep, above, ran longer than a
360 // full second, at which point we assume sockets may have been killed (they
361 // appear to be at least on some platforms, even if it has only been a second).
362 // Note that we have to take care to not get here just because user event
363 // processing was slow at the top of the loop. For example, the sample client
364 // may call Bitcoin Core RPCs during event handling, which very often takes
365 // more than a handful of seconds to complete, and shouldn't disconnect all our
367 log_trace!($logger, "100ms sleep took more than a second, disconnecting peers.");
368 $peer_manager.as_ref().disconnect_all_peers();
369 last_ping_call = $get_timer(PING_TIMER);
370 } else if $timer_elapsed(&mut last_ping_call, PING_TIMER) {
371 log_trace!($logger, "Calling PeerManager's timer_tick_occurred");
372 $peer_manager.as_ref().timer_tick_occurred();
373 last_ping_call = $get_timer(PING_TIMER);
376 // Note that we want to run a graph prune once not long after startup before
377 // falling back to our usual hourly prunes. This avoids short-lived clients never
378 // pruning their network graph. We run once 60 seconds after startup before
379 // continuing our normal cadence. For RGS, since 60 seconds is likely too long,
380 // we prune after an initial sync completes.
381 let prune_timer = if have_pruned { NETWORK_PRUNE_TIMER } else { FIRST_NETWORK_PRUNE_TIMER };
382 let prune_timer_elapsed = $timer_elapsed(&mut last_prune_call, prune_timer);
383 let should_prune = match $gossip_sync {
384 GossipSync::Rapid(_) => !have_pruned || prune_timer_elapsed,
385 _ => prune_timer_elapsed,
388 // The network graph must not be pruned while rapid sync completion is pending
389 if let Some(network_graph) = $gossip_sync.prunable_network_graph() {
390 if let Some(duration_since_epoch) = $time_fetch() {
391 log_trace!($logger, "Pruning and persisting network graph.");
392 network_graph.remove_stale_channels_and_tracking_with_time(duration_since_epoch.as_secs());
394 log_warn!($logger, "Not pruning network graph, consider enabling `std` or doing so manually with remove_stale_channels_and_tracking_with_time.");
395 log_trace!($logger, "Persisting network graph.");
398 if let Err(e) = $persister.persist_graph(network_graph) {
399 log_error!($logger, "Error: Failed to persist network graph, check your disk and permissions {}", e)
404 let prune_timer = if have_pruned { NETWORK_PRUNE_TIMER } else { FIRST_NETWORK_PRUNE_TIMER };
405 last_prune_call = $get_timer(prune_timer);
408 if !have_decayed_scorer {
409 if let Some(ref scorer) = $scorer {
410 if let Some(duration_since_epoch) = $time_fetch() {
411 log_trace!($logger, "Calling time_passed on scorer at startup");
412 scorer.write_lock().time_passed(duration_since_epoch);
415 have_decayed_scorer = true;
418 if $timer_elapsed(&mut last_scorer_persist_call, SCORER_PERSIST_TIMER) {
419 if let Some(ref scorer) = $scorer {
420 if let Some(duration_since_epoch) = $time_fetch() {
421 log_trace!($logger, "Calling time_passed and persisting scorer");
422 scorer.write_lock().time_passed(duration_since_epoch);
424 log_trace!($logger, "Persisting scorer");
426 if let Err(e) = $persister.persist_scorer(&scorer) {
427 log_error!($logger, "Error: Failed to persist scorer, check your disk and permissions {}", e)
430 last_scorer_persist_call = $get_timer(SCORER_PERSIST_TIMER);
433 if $timer_elapsed(&mut last_rebroadcast_call, REBROADCAST_TIMER) {
434 log_trace!($logger, "Rebroadcasting monitor's pending claims");
435 $chain_monitor.rebroadcast_pending_claims();
436 last_rebroadcast_call = $get_timer(REBROADCAST_TIMER);
440 // After we exit, ensure we persist the ChannelManager one final time - this avoids
441 // some races where users quit while channel updates were in-flight, with
442 // ChannelMonitor update(s) persisted without a corresponding ChannelManager update.
443 $persister.persist_manager(&$channel_manager)?;
445 // Persist Scorer on exit
446 if let Some(ref scorer) = $scorer {
447 $persister.persist_scorer(&scorer)?;
450 // Persist NetworkGraph on exit
451 if let Some(network_graph) = $gossip_sync.network_graph() {
452 $persister.persist_graph(network_graph)?;
459 #[cfg(feature = "futures")]
460 pub(crate) mod futures_util {
461 use core::future::Future;
462 use core::task::{Poll, Waker, RawWaker, RawWakerVTable};
464 use core::marker::Unpin;
465 pub(crate) struct Selector<
466 A: Future<Output=()> + Unpin, B: Future<Output=()> + Unpin, C: Future<Output=bool> + Unpin
472 pub(crate) enum SelectorOutput {
477 A: Future<Output=()> + Unpin, B: Future<Output=()> + Unpin, C: Future<Output=bool> + Unpin
478 > Future for Selector<A, B, C> {
479 type Output = SelectorOutput;
480 fn poll(mut self: Pin<&mut Self>, ctx: &mut core::task::Context<'_>) -> Poll<SelectorOutput> {
481 match Pin::new(&mut self.a).poll(ctx) {
482 Poll::Ready(()) => { return Poll::Ready(SelectorOutput::A); },
485 match Pin::new(&mut self.b).poll(ctx) {
486 Poll::Ready(()) => { return Poll::Ready(SelectorOutput::B); },
489 match Pin::new(&mut self.c).poll(ctx) {
490 Poll::Ready(res) => { return Poll::Ready(SelectorOutput::C(res)); },
497 // If we want to poll a future without an async context to figure out if it has completed or
498 // not without awaiting, we need a Waker, which needs a vtable...we fill it with dummy values
499 // but sadly there's a good bit of boilerplate here.
500 fn dummy_waker_clone(_: *const ()) -> RawWaker { RawWaker::new(core::ptr::null(), &DUMMY_WAKER_VTABLE) }
501 fn dummy_waker_action(_: *const ()) { }
503 const DUMMY_WAKER_VTABLE: RawWakerVTable = RawWakerVTable::new(
504 dummy_waker_clone, dummy_waker_action, dummy_waker_action, dummy_waker_action);
505 pub(crate) fn dummy_waker() -> Waker { unsafe { Waker::from_raw(RawWaker::new(core::ptr::null(), &DUMMY_WAKER_VTABLE)) } }
507 #[cfg(feature = "futures")]
508 use futures_util::{Selector, SelectorOutput, dummy_waker};
509 #[cfg(feature = "futures")]
512 /// Processes background events in a future.
514 /// `sleeper` should return a future which completes in the given amount of time and returns a
515 /// boolean indicating whether the background processing should exit. Once `sleeper` returns a
516 /// future which outputs `true`, the loop will exit and this function's future will complete.
517 /// The `sleeper` future is free to return early after it has triggered the exit condition.
519 /// See [`BackgroundProcessor::start`] for information on which actions this handles.
521 /// Requires the `futures` feature. Note that while this method is available without the `std`
522 /// feature, doing so will skip calling [`NetworkGraph::remove_stale_channels_and_tracking`],
523 /// you should call [`NetworkGraph::remove_stale_channels_and_tracking_with_time`] regularly
524 /// manually instead.
526 /// The `mobile_interruptable_platform` flag should be set if we're currently running on a
527 /// mobile device, where we may need to check for interruption of the application regularly. If you
528 /// are unsure, you should set the flag, as the performance impact of it is minimal unless there
529 /// are hundreds or thousands of simultaneous process calls running.
531 /// The `fetch_time` parameter should return the current wall clock time, if one is available. If
532 /// no time is available, some features may be disabled, however the node will still operate fine.
534 /// For example, in order to process background events in a [Tokio](https://tokio.rs/) task, you
535 /// could setup `process_events_async` like this:
537 /// # use lightning::io;
538 /// # use std::sync::{Arc, RwLock};
539 /// # use std::sync::atomic::{AtomicBool, Ordering};
540 /// # use std::time::SystemTime;
541 /// # use lightning_background_processor::{process_events_async, GossipSync};
542 /// # struct Logger {}
543 /// # impl lightning::util::logger::Logger for Logger {
544 /// # fn log(&self, _record: lightning::util::logger::Record) {}
546 /// # struct Store {}
547 /// # impl lightning::util::persist::KVStore for Store {
548 /// # fn read(&self, primary_namespace: &str, secondary_namespace: &str, key: &str) -> io::Result<Vec<u8>> { Ok(Vec::new()) }
549 /// # fn write(&self, primary_namespace: &str, secondary_namespace: &str, key: &str, buf: &[u8]) -> io::Result<()> { Ok(()) }
550 /// # fn remove(&self, primary_namespace: &str, secondary_namespace: &str, key: &str, lazy: bool) -> io::Result<()> { Ok(()) }
551 /// # fn list(&self, primary_namespace: &str, secondary_namespace: &str) -> io::Result<Vec<String>> { Ok(Vec::new()) }
553 /// # struct EventHandler {}
554 /// # impl EventHandler {
555 /// # async fn handle_event(&self, _: lightning::events::Event) {}
557 /// # #[derive(Eq, PartialEq, Clone, Hash)]
558 /// # struct SocketDescriptor {}
559 /// # impl lightning::ln::peer_handler::SocketDescriptor for SocketDescriptor {
560 /// # fn send_data(&mut self, _data: &[u8], _resume_read: bool) -> usize { 0 }
561 /// # fn disconnect_socket(&mut self) {}
563 /// # type ChainMonitor<B, F, FE> = lightning::chain::chainmonitor::ChainMonitor<lightning::sign::InMemorySigner, Arc<F>, Arc<B>, Arc<FE>, Arc<Logger>, Arc<Store>>;
564 /// # type NetworkGraph = lightning::routing::gossip::NetworkGraph<Arc<Logger>>;
565 /// # type P2PGossipSync<UL> = lightning::routing::gossip::P2PGossipSync<Arc<NetworkGraph>, Arc<UL>, Arc<Logger>>;
566 /// # type ChannelManager<B, F, FE> = lightning::ln::channelmanager::SimpleArcChannelManager<ChainMonitor<B, F, FE>, B, FE, Logger>;
567 /// # type Scorer = RwLock<lightning::routing::scoring::ProbabilisticScorer<Arc<NetworkGraph>, Arc<Logger>>>;
568 /// # type PeerManager<B, F, FE, UL> = lightning::ln::peer_handler::SimpleArcPeerManager<SocketDescriptor, ChainMonitor<B, F, FE>, B, FE, Arc<UL>, Logger>;
571 /// # B: lightning::chain::chaininterface::BroadcasterInterface + Send + Sync + 'static,
572 /// # F: lightning::chain::Filter + Send + Sync + 'static,
573 /// # FE: lightning::chain::chaininterface::FeeEstimator + Send + Sync + 'static,
574 /// # UL: lightning::routing::utxo::UtxoLookup + Send + Sync + 'static,
576 /// # peer_manager: Arc<PeerManager<B, F, FE, UL>>,
577 /// # event_handler: Arc<EventHandler>,
578 /// # channel_manager: Arc<ChannelManager<B, F, FE>>,
579 /// # chain_monitor: Arc<ChainMonitor<B, F, FE>>,
580 /// # gossip_sync: Arc<P2PGossipSync<UL>>,
581 /// # persister: Arc<Store>,
582 /// # logger: Arc<Logger>,
583 /// # scorer: Arc<Scorer>,
586 /// # async fn setup_background_processing<
587 /// # B: lightning::chain::chaininterface::BroadcasterInterface + Send + Sync + 'static,
588 /// # F: lightning::chain::Filter + Send + Sync + 'static,
589 /// # FE: lightning::chain::chaininterface::FeeEstimator + Send + Sync + 'static,
590 /// # UL: lightning::routing::utxo::UtxoLookup + Send + Sync + 'static,
591 /// # >(node: Node<B, F, FE, UL>) {
592 /// let background_persister = Arc::clone(&node.persister);
593 /// let background_event_handler = Arc::clone(&node.event_handler);
594 /// let background_chain_mon = Arc::clone(&node.chain_monitor);
595 /// let background_chan_man = Arc::clone(&node.channel_manager);
596 /// let background_gossip_sync = GossipSync::p2p(Arc::clone(&node.gossip_sync));
597 /// let background_peer_man = Arc::clone(&node.peer_manager);
598 /// let background_logger = Arc::clone(&node.logger);
599 /// let background_scorer = Arc::clone(&node.scorer);
601 /// // Setup the sleeper.
602 /// let (stop_sender, stop_receiver) = tokio::sync::watch::channel(());
604 /// let sleeper = move |d| {
605 /// let mut receiver = stop_receiver.clone();
606 /// Box::pin(async move {
608 /// _ = tokio::time::sleep(d) => false,
609 /// _ = receiver.changed() => true,
614 /// let mobile_interruptable_platform = false;
616 /// let handle = tokio::spawn(async move {
617 /// process_events_async(
618 /// background_persister,
619 /// |e| background_event_handler.handle_event(e),
620 /// background_chain_mon,
621 /// background_chan_man,
622 /// background_gossip_sync,
623 /// background_peer_man,
624 /// background_logger,
625 /// Some(background_scorer),
627 /// mobile_interruptable_platform,
628 /// || Some(SystemTime::now().duration_since(SystemTime::UNIX_EPOCH).unwrap())
631 /// .expect("Failed to process events");
634 /// // Stop the background processing.
635 /// stop_sender.send(()).unwrap();
636 /// handle.await.unwrap();
639 #[cfg(feature = "futures")]
640 pub async fn process_events_async<
642 UL: 'static + Deref + Send + Sync,
643 CF: 'static + Deref + Send + Sync,
644 T: 'static + Deref + Send + Sync,
645 F: 'static + Deref + Send + Sync,
646 G: 'static + Deref<Target = NetworkGraph<L>> + Send + Sync,
647 L: 'static + Deref + Send + Sync,
648 P: 'static + Deref + Send + Sync,
649 EventHandlerFuture: core::future::Future<Output = ()>,
650 EventHandler: Fn(Event) -> EventHandlerFuture,
651 PS: 'static + Deref + Send,
652 M: 'static + Deref<Target = ChainMonitor<<CM::Target as AChannelManager>::Signer, CF, T, F, L, P>> + Send + Sync,
653 CM: 'static + Deref + Send + Sync,
654 PGS: 'static + Deref<Target = P2PGossipSync<G, UL, L>> + Send + Sync,
655 RGS: 'static + Deref<Target = RapidGossipSync<G, L>> + Send,
656 PM: 'static + Deref + Send + Sync,
657 S: 'static + Deref<Target = SC> + Send + Sync,
658 SC: for<'b> WriteableScore<'b>,
659 SleepFuture: core::future::Future<Output = bool> + core::marker::Unpin,
660 Sleeper: Fn(Duration) -> SleepFuture,
661 FetchTime: Fn() -> Option<Duration>,
663 persister: PS, event_handler: EventHandler, chain_monitor: M, channel_manager: CM,
664 gossip_sync: GossipSync<PGS, RGS, G, UL, L>, peer_manager: PM, logger: L, scorer: Option<S>,
665 sleeper: Sleeper, mobile_interruptable_platform: bool, fetch_time: FetchTime,
666 ) -> Result<(), lightning::io::Error>
668 UL::Target: 'static + UtxoLookup,
669 CF::Target: 'static + chain::Filter,
670 T::Target: 'static + BroadcasterInterface,
671 F::Target: 'static + FeeEstimator,
672 L::Target: 'static + Logger,
673 P::Target: 'static + Persist<<CM::Target as AChannelManager>::Signer>,
674 PS::Target: 'static + Persister<'a, CM, L, SC>,
675 CM::Target: AChannelManager + Send + Sync,
676 PM::Target: APeerManager + Send + Sync,
678 let mut should_break = false;
679 let async_event_handler = |event| {
680 let network_graph = gossip_sync.network_graph();
681 let event_handler = &event_handler;
682 let scorer = &scorer;
683 let logger = &logger;
684 let persister = &persister;
685 let fetch_time = &fetch_time;
687 if let Some(network_graph) = network_graph {
688 handle_network_graph_update(network_graph, &event)
690 if let Some(ref scorer) = scorer {
691 if let Some(duration_since_epoch) = fetch_time() {
692 if update_scorer(scorer, &event, duration_since_epoch) {
693 log_trace!(logger, "Persisting scorer after update");
694 if let Err(e) = persister.persist_scorer(&scorer) {
695 log_error!(logger, "Error: Failed to persist scorer, check your disk and permissions {}", e)
700 event_handler(event).await;
704 persister, chain_monitor,
705 chain_monitor.process_pending_events_async(async_event_handler).await,
706 channel_manager, channel_manager.get_cm().process_pending_events_async(async_event_handler).await,
707 peer_manager, process_onion_message_handler_events_async(&peer_manager, async_event_handler).await,
708 gossip_sync, logger, scorer, should_break, {
710 a: channel_manager.get_cm().get_event_or_persistence_needed_future(),
711 b: chain_monitor.get_update_future(),
712 c: sleeper(if mobile_interruptable_platform { Duration::from_millis(100) } else { Duration::from_secs(FASTEST_TIMER) }),
715 SelectorOutput::A|SelectorOutput::B => {},
716 SelectorOutput::C(exit) => {
720 }, |t| sleeper(Duration::from_secs(t)),
721 |fut: &mut SleepFuture, _| {
722 let mut waker = dummy_waker();
723 let mut ctx = task::Context::from_waker(&mut waker);
724 match core::pin::Pin::new(fut).poll(&mut ctx) {
725 task::Poll::Ready(exit) => { should_break = exit; true },
726 task::Poll::Pending => false,
728 }, mobile_interruptable_platform, fetch_time,
732 #[cfg(feature = "futures")]
733 async fn process_onion_message_handler_events_async<
734 EventHandlerFuture: core::future::Future<Output = ()>,
735 EventHandler: Fn(Event) -> EventHandlerFuture,
736 PM: 'static + Deref + Send + Sync,
738 peer_manager: &PM, handler: EventHandler
741 PM::Target: APeerManager + Send + Sync,
743 let events = core::cell::RefCell::new(Vec::new());
744 peer_manager.onion_message_handler().process_pending_events(&|e| events.borrow_mut().push(e));
746 for event in events.into_inner() {
751 #[cfg(feature = "std")]
752 impl BackgroundProcessor {
753 /// Start a background thread that takes care of responsibilities enumerated in the [top-level
756 /// The thread runs indefinitely unless the object is dropped, [`stop`] is called, or
757 /// [`Persister::persist_manager`] returns an error. In case of an error, the error is retrieved by calling
758 /// either [`join`] or [`stop`].
760 /// # Data Persistence
762 /// [`Persister::persist_manager`] is responsible for writing out the [`ChannelManager`] to disk, and/or
763 /// uploading to one or more backup services. See [`ChannelManager::write`] for writing out a
764 /// [`ChannelManager`]. See the `lightning-persister` crate for LDK's
765 /// provided implementation.
767 /// [`Persister::persist_graph`] is responsible for writing out the [`NetworkGraph`] to disk, if
768 /// [`GossipSync`] is supplied. See [`NetworkGraph::write`] for writing out a [`NetworkGraph`].
769 /// See the `lightning-persister` crate for LDK's provided implementation.
771 /// Typically, users should either implement [`Persister::persist_manager`] to never return an
772 /// error or call [`join`] and handle any error that may arise. For the latter case,
773 /// `BackgroundProcessor` must be restarted by calling `start` again after handling the error.
777 /// `event_handler` is responsible for handling events that users should be notified of (e.g.,
778 /// payment failed). [`BackgroundProcessor`] may decorate the given [`EventHandler`] with common
779 /// functionality implemented by other handlers.
780 /// * [`P2PGossipSync`] if given will update the [`NetworkGraph`] based on payment failures.
782 /// # Rapid Gossip Sync
784 /// If rapid gossip sync is meant to run at startup, pass [`RapidGossipSync`] via `gossip_sync`
785 /// to indicate that the [`BackgroundProcessor`] should not prune the [`NetworkGraph`] instance
786 /// until the [`RapidGossipSync`] instance completes its first sync.
788 /// [top-level documentation]: BackgroundProcessor
789 /// [`join`]: Self::join
790 /// [`stop`]: Self::stop
791 /// [`ChannelManager`]: lightning::ln::channelmanager::ChannelManager
792 /// [`ChannelManager::write`]: lightning::ln::channelmanager::ChannelManager#impl-Writeable
793 /// [`Persister::persist_manager`]: lightning::util::persist::Persister::persist_manager
794 /// [`Persister::persist_graph`]: lightning::util::persist::Persister::persist_graph
795 /// [`NetworkGraph`]: lightning::routing::gossip::NetworkGraph
796 /// [`NetworkGraph::write`]: lightning::routing::gossip::NetworkGraph#impl-Writeable
799 UL: 'static + Deref + Send + Sync,
800 CF: 'static + Deref + Send + Sync,
801 T: 'static + Deref + Send + Sync,
802 F: 'static + Deref + Send + Sync,
803 G: 'static + Deref<Target = NetworkGraph<L>> + Send + Sync,
804 L: 'static + Deref + Send + Sync,
805 P: 'static + Deref + Send + Sync,
806 EH: 'static + EventHandler + Send,
807 PS: 'static + Deref + Send,
808 M: 'static + Deref<Target = ChainMonitor<<CM::Target as AChannelManager>::Signer, CF, T, F, L, P>> + Send + Sync,
809 CM: 'static + Deref + Send + Sync,
810 PGS: 'static + Deref<Target = P2PGossipSync<G, UL, L>> + Send + Sync,
811 RGS: 'static + Deref<Target = RapidGossipSync<G, L>> + Send,
812 PM: 'static + Deref + Send + Sync,
813 S: 'static + Deref<Target = SC> + Send + Sync,
814 SC: for <'b> WriteableScore<'b>,
816 persister: PS, event_handler: EH, chain_monitor: M, channel_manager: CM,
817 gossip_sync: GossipSync<PGS, RGS, G, UL, L>, peer_manager: PM, logger: L, scorer: Option<S>,
820 UL::Target: 'static + UtxoLookup,
821 CF::Target: 'static + chain::Filter,
822 T::Target: 'static + BroadcasterInterface,
823 F::Target: 'static + FeeEstimator,
824 L::Target: 'static + Logger,
825 P::Target: 'static + Persist<<CM::Target as AChannelManager>::Signer>,
826 PS::Target: 'static + Persister<'a, CM, L, SC>,
827 CM::Target: AChannelManager + Send + Sync,
828 PM::Target: APeerManager + Send + Sync,
830 let stop_thread = Arc::new(AtomicBool::new(false));
831 let stop_thread_clone = stop_thread.clone();
832 let handle = thread::spawn(move || -> Result<(), std::io::Error> {
833 let event_handler = |event| {
834 let network_graph = gossip_sync.network_graph();
835 if let Some(network_graph) = network_graph {
836 handle_network_graph_update(network_graph, &event)
838 if let Some(ref scorer) = scorer {
839 use std::time::SystemTime;
840 let duration_since_epoch = SystemTime::now().duration_since(SystemTime::UNIX_EPOCH)
841 .expect("Time should be sometime after 1970");
842 if update_scorer(scorer, &event, duration_since_epoch) {
843 log_trace!(logger, "Persisting scorer after update");
844 if let Err(e) = persister.persist_scorer(&scorer) {
845 log_error!(logger, "Error: Failed to persist scorer, check your disk and permissions {}", e)
849 event_handler.handle_event(event);
852 persister, chain_monitor, chain_monitor.process_pending_events(&event_handler),
853 channel_manager, channel_manager.get_cm().process_pending_events(&event_handler),
855 peer_manager.onion_message_handler().process_pending_events(&event_handler),
856 gossip_sync, logger, scorer, stop_thread.load(Ordering::Acquire),
857 { Sleeper::from_two_futures(
858 &channel_manager.get_cm().get_event_or_persistence_needed_future(),
859 &chain_monitor.get_update_future()
860 ).wait_timeout(Duration::from_millis(100)); },
861 |_| Instant::now(), |time: &Instant, dur| time.elapsed().as_secs() > dur, false,
863 use std::time::SystemTime;
864 Some(SystemTime::now().duration_since(SystemTime::UNIX_EPOCH)
865 .expect("Time should be sometime after 1970"))
869 Self { stop_thread: stop_thread_clone, thread_handle: Some(handle) }
872 /// Join `BackgroundProcessor`'s thread, returning any error that occurred while persisting
873 /// [`ChannelManager`].
877 /// This function panics if the background thread has panicked such as while persisting or
880 /// [`ChannelManager`]: lightning::ln::channelmanager::ChannelManager
881 pub fn join(mut self) -> Result<(), std::io::Error> {
882 assert!(self.thread_handle.is_some());
886 /// Stop `BackgroundProcessor`'s thread, returning any error that occurred while persisting
887 /// [`ChannelManager`].
891 /// This function panics if the background thread has panicked such as while persisting or
894 /// [`ChannelManager`]: lightning::ln::channelmanager::ChannelManager
895 pub fn stop(mut self) -> Result<(), std::io::Error> {
896 assert!(self.thread_handle.is_some());
897 self.stop_and_join_thread()
900 fn stop_and_join_thread(&mut self) -> Result<(), std::io::Error> {
901 self.stop_thread.store(true, Ordering::Release);
905 fn join_thread(&mut self) -> Result<(), std::io::Error> {
906 match self.thread_handle.take() {
907 Some(handle) => handle.join().unwrap(),
913 #[cfg(feature = "std")]
914 impl Drop for BackgroundProcessor {
916 self.stop_and_join_thread().unwrap();
920 #[cfg(all(feature = "std", test))]
922 use bitcoin::{Amount, ScriptBuf, Txid};
923 use bitcoin::blockdata::constants::{genesis_block, ChainHash};
924 use bitcoin::blockdata::locktime::absolute::LockTime;
925 use bitcoin::blockdata::transaction::{Transaction, TxOut};
926 use bitcoin::hashes::Hash;
927 use bitcoin::network::Network;
928 use bitcoin::secp256k1::{SecretKey, PublicKey, Secp256k1};
929 use bitcoin::transaction::Version;
930 use lightning::chain::{BestBlock, Confirm, chainmonitor, Filter};
931 use lightning::chain::channelmonitor::ANTI_REORG_DELAY;
932 use lightning::sign::{InMemorySigner, KeysManager, ChangeDestinationSource};
933 use lightning::chain::transaction::OutPoint;
934 use lightning::events::{Event, PathFailure, MessageSendEventsProvider, MessageSendEvent};
935 use lightning::{get_event_msg, get_event};
936 use lightning::ln::types::{PaymentHash, ChannelId};
937 use lightning::ln::channelmanager;
938 use lightning::ln::channelmanager::{BREAKDOWN_TIMEOUT, ChainParameters, MIN_CLTV_EXPIRY_DELTA, PaymentId};
939 use lightning::ln::features::{ChannelFeatures, NodeFeatures};
940 use lightning::ln::functional_test_utils::*;
941 use lightning::ln::msgs::{ChannelMessageHandler, Init};
942 use lightning::ln::peer_handler::{PeerManager, MessageHandler, SocketDescriptor, IgnoringMessageHandler};
943 use lightning::routing::gossip::{NetworkGraph, P2PGossipSync};
944 use lightning::routing::scoring::{ChannelUsage, ScoreUpdate, ScoreLookUp, LockableScore};
945 use lightning::routing::router::{DefaultRouter, Path, RouteHop, CandidateRouteHop};
946 use lightning::util::config::UserConfig;
947 use lightning::util::ser::Writeable;
948 use lightning::util::test_utils;
949 use lightning::util::persist::{KVStore,
950 CHANNEL_MANAGER_PERSISTENCE_PRIMARY_NAMESPACE, CHANNEL_MANAGER_PERSISTENCE_SECONDARY_NAMESPACE, CHANNEL_MANAGER_PERSISTENCE_KEY,
951 NETWORK_GRAPH_PERSISTENCE_PRIMARY_NAMESPACE, NETWORK_GRAPH_PERSISTENCE_SECONDARY_NAMESPACE, NETWORK_GRAPH_PERSISTENCE_KEY,
952 SCORER_PERSISTENCE_PRIMARY_NAMESPACE, SCORER_PERSISTENCE_SECONDARY_NAMESPACE, SCORER_PERSISTENCE_KEY};
953 use lightning::util::sweep::{OutputSweeper, OutputSpendStatus};
954 use lightning_persister::fs_store::FilesystemStore;
955 use std::collections::VecDeque;
957 use std::path::PathBuf;
958 use std::sync::{Arc, Mutex};
959 use std::sync::mpsc::SyncSender;
960 use std::time::Duration;
961 use lightning_rapid_gossip_sync::RapidGossipSync;
962 use super::{BackgroundProcessor, GossipSync, FRESHNESS_TIMER};
964 const EVENT_DEADLINE: u64 = 5 * FRESHNESS_TIMER;
966 #[derive(Clone, Hash, PartialEq, Eq)]
967 struct TestDescriptor{}
968 impl SocketDescriptor for TestDescriptor {
969 fn send_data(&mut self, _data: &[u8], _resume_read: bool) -> usize {
973 fn disconnect_socket(&mut self) {}
977 type LockingWrapper<T> = lightning::routing::scoring::MultiThreadedLockableScore<T>;
978 #[cfg(not(c_bindings))]
979 type LockingWrapper<T> = Mutex<T>;
981 type ChannelManager =
982 channelmanager::ChannelManager<
984 Arc<test_utils::TestBroadcaster>,
988 Arc<test_utils::TestFeeEstimator>,
990 Arc<NetworkGraph<Arc<test_utils::TestLogger>>>,
991 Arc<test_utils::TestLogger>,
993 Arc<LockingWrapper<TestScorer>>,
997 Arc<test_utils::TestLogger>>;
999 type ChainMonitor = chainmonitor::ChainMonitor<InMemorySigner, Arc<test_utils::TestChainSource>, Arc<test_utils::TestBroadcaster>, Arc<test_utils::TestFeeEstimator>, Arc<test_utils::TestLogger>, Arc<FilesystemStore>>;
1001 type PGS = Arc<P2PGossipSync<Arc<NetworkGraph<Arc<test_utils::TestLogger>>>, Arc<test_utils::TestChainSource>, Arc<test_utils::TestLogger>>>;
1002 type RGS = Arc<RapidGossipSync<Arc<NetworkGraph<Arc<test_utils::TestLogger>>>, Arc<test_utils::TestLogger>>>;
1005 node: Arc<ChannelManager>,
1006 p2p_gossip_sync: PGS,
1007 rapid_gossip_sync: RGS,
1008 peer_manager: Arc<PeerManager<TestDescriptor, Arc<test_utils::TestChannelMessageHandler>, Arc<test_utils::TestRoutingMessageHandler>, IgnoringMessageHandler, Arc<test_utils::TestLogger>, IgnoringMessageHandler, Arc<KeysManager>>>,
1009 chain_monitor: Arc<ChainMonitor>,
1010 kv_store: Arc<FilesystemStore>,
1011 tx_broadcaster: Arc<test_utils::TestBroadcaster>,
1012 network_graph: Arc<NetworkGraph<Arc<test_utils::TestLogger>>>,
1013 logger: Arc<test_utils::TestLogger>,
1014 best_block: BestBlock,
1015 scorer: Arc<LockingWrapper<TestScorer>>,
1016 sweeper: Arc<OutputSweeper<Arc<test_utils::TestBroadcaster>, Arc<TestWallet>,
1017 Arc<test_utils::TestFeeEstimator>, Arc<dyn Filter + Sync + Send>, Arc<FilesystemStore>,
1018 Arc<test_utils::TestLogger>, Arc<KeysManager>>>,
1022 fn p2p_gossip_sync(&self) -> GossipSync<PGS, RGS, Arc<NetworkGraph<Arc<test_utils::TestLogger>>>, Arc<test_utils::TestChainSource>, Arc<test_utils::TestLogger>> {
1023 GossipSync::P2P(self.p2p_gossip_sync.clone())
1026 fn rapid_gossip_sync(&self) -> GossipSync<PGS, RGS, Arc<NetworkGraph<Arc<test_utils::TestLogger>>>, Arc<test_utils::TestChainSource>, Arc<test_utils::TestLogger>> {
1027 GossipSync::Rapid(self.rapid_gossip_sync.clone())
1030 fn no_gossip_sync(&self) -> GossipSync<PGS, RGS, Arc<NetworkGraph<Arc<test_utils::TestLogger>>>, Arc<test_utils::TestChainSource>, Arc<test_utils::TestLogger>> {
1035 impl Drop for Node {
1036 fn drop(&mut self) {
1037 let data_dir = self.kv_store.get_data_dir();
1038 match fs::remove_dir_all(data_dir.clone()) {
1039 Err(e) => println!("Failed to remove test store directory {}: {}", data_dir.display(), e),
1046 graph_error: Option<(std::io::ErrorKind, &'static str)>,
1047 graph_persistence_notifier: Option<SyncSender<()>>,
1048 manager_error: Option<(std::io::ErrorKind, &'static str)>,
1049 scorer_error: Option<(std::io::ErrorKind, &'static str)>,
1050 kv_store: FilesystemStore,
1054 fn new(data_dir: PathBuf) -> Self {
1055 let kv_store = FilesystemStore::new(data_dir);
1056 Self { graph_error: None, graph_persistence_notifier: None, manager_error: None, scorer_error: None, kv_store }
1059 fn with_graph_error(self, error: std::io::ErrorKind, message: &'static str) -> Self {
1060 Self { graph_error: Some((error, message)), ..self }
1063 fn with_graph_persistence_notifier(self, sender: SyncSender<()>) -> Self {
1064 Self { graph_persistence_notifier: Some(sender), ..self }
1067 fn with_manager_error(self, error: std::io::ErrorKind, message: &'static str) -> Self {
1068 Self { manager_error: Some((error, message)), ..self }
1071 fn with_scorer_error(self, error: std::io::ErrorKind, message: &'static str) -> Self {
1072 Self { scorer_error: Some((error, message)), ..self }
1076 impl KVStore for Persister {
1077 fn read(&self, primary_namespace: &str, secondary_namespace: &str, key: &str) -> lightning::io::Result<Vec<u8>> {
1078 self.kv_store.read(primary_namespace, secondary_namespace, key)
1081 fn write(&self, primary_namespace: &str, secondary_namespace: &str, key: &str, buf: &[u8]) -> lightning::io::Result<()> {
1082 if primary_namespace == CHANNEL_MANAGER_PERSISTENCE_PRIMARY_NAMESPACE &&
1083 secondary_namespace == CHANNEL_MANAGER_PERSISTENCE_SECONDARY_NAMESPACE &&
1084 key == CHANNEL_MANAGER_PERSISTENCE_KEY
1086 if let Some((error, message)) = self.manager_error {
1087 return Err(std::io::Error::new(error, message))
1091 if primary_namespace == NETWORK_GRAPH_PERSISTENCE_PRIMARY_NAMESPACE &&
1092 secondary_namespace == NETWORK_GRAPH_PERSISTENCE_SECONDARY_NAMESPACE &&
1093 key == NETWORK_GRAPH_PERSISTENCE_KEY
1095 if let Some(sender) = &self.graph_persistence_notifier {
1096 match sender.send(()) {
1098 Err(std::sync::mpsc::SendError(())) => println!("Persister failed to notify as receiver went away."),
1102 if let Some((error, message)) = self.graph_error {
1103 return Err(std::io::Error::new(error, message))
1107 if primary_namespace == SCORER_PERSISTENCE_PRIMARY_NAMESPACE &&
1108 secondary_namespace == SCORER_PERSISTENCE_SECONDARY_NAMESPACE &&
1109 key == SCORER_PERSISTENCE_KEY
1111 if let Some((error, message)) = self.scorer_error {
1112 return Err(std::io::Error::new(error, message))
1116 self.kv_store.write(primary_namespace, secondary_namespace, key, buf)
1119 fn remove(&self, primary_namespace: &str, secondary_namespace: &str, key: &str, lazy: bool) -> lightning::io::Result<()> {
1120 self.kv_store.remove(primary_namespace, secondary_namespace, key, lazy)
1123 fn list(&self, primary_namespace: &str, secondary_namespace: &str) -> lightning::io::Result<Vec<String>> {
1124 self.kv_store.list(primary_namespace, secondary_namespace)
1129 event_expectations: Option<VecDeque<TestResult>>,
1134 PaymentFailure { path: Path, short_channel_id: u64 },
1135 PaymentSuccess { path: Path },
1136 ProbeFailure { path: Path },
1137 ProbeSuccess { path: Path },
1142 Self { event_expectations: None }
1145 fn expect(&mut self, expectation: TestResult) {
1146 self.event_expectations.get_or_insert_with(VecDeque::new).push_back(expectation);
1150 impl lightning::util::ser::Writeable for TestScorer {
1151 fn write<W: lightning::util::ser::Writer>(&self, _: &mut W) -> Result<(), lightning::io::Error> { Ok(()) }
1154 impl ScoreLookUp for TestScorer {
1155 type ScoreParams = ();
1156 fn channel_penalty_msat(
1157 &self, _candidate: &CandidateRouteHop, _usage: ChannelUsage, _score_params: &Self::ScoreParams
1158 ) -> u64 { unimplemented!(); }
1161 impl ScoreUpdate for TestScorer {
1162 fn payment_path_failed(&mut self, actual_path: &Path, actual_short_channel_id: u64, _: Duration) {
1163 if let Some(expectations) = &mut self.event_expectations {
1164 match expectations.pop_front().unwrap() {
1165 TestResult::PaymentFailure { path, short_channel_id } => {
1166 assert_eq!(actual_path, &path);
1167 assert_eq!(actual_short_channel_id, short_channel_id);
1169 TestResult::PaymentSuccess { path } => {
1170 panic!("Unexpected successful payment path: {:?}", path)
1172 TestResult::ProbeFailure { path } => {
1173 panic!("Unexpected probe failure: {:?}", path)
1175 TestResult::ProbeSuccess { path } => {
1176 panic!("Unexpected probe success: {:?}", path)
1182 fn payment_path_successful(&mut self, actual_path: &Path, _: Duration) {
1183 if let Some(expectations) = &mut self.event_expectations {
1184 match expectations.pop_front().unwrap() {
1185 TestResult::PaymentFailure { path, .. } => {
1186 panic!("Unexpected payment path failure: {:?}", path)
1188 TestResult::PaymentSuccess { path } => {
1189 assert_eq!(actual_path, &path);
1191 TestResult::ProbeFailure { path } => {
1192 panic!("Unexpected probe failure: {:?}", path)
1194 TestResult::ProbeSuccess { path } => {
1195 panic!("Unexpected probe success: {:?}", path)
1201 fn probe_failed(&mut self, actual_path: &Path, _: u64, _: Duration) {
1202 if let Some(expectations) = &mut self.event_expectations {
1203 match expectations.pop_front().unwrap() {
1204 TestResult::PaymentFailure { path, .. } => {
1205 panic!("Unexpected payment path failure: {:?}", path)
1207 TestResult::PaymentSuccess { path } => {
1208 panic!("Unexpected payment path success: {:?}", path)
1210 TestResult::ProbeFailure { path } => {
1211 assert_eq!(actual_path, &path);
1213 TestResult::ProbeSuccess { path } => {
1214 panic!("Unexpected probe success: {:?}", path)
1219 fn probe_successful(&mut self, actual_path: &Path, _: Duration) {
1220 if let Some(expectations) = &mut self.event_expectations {
1221 match expectations.pop_front().unwrap() {
1222 TestResult::PaymentFailure { path, .. } => {
1223 panic!("Unexpected payment path failure: {:?}", path)
1225 TestResult::PaymentSuccess { path } => {
1226 panic!("Unexpected payment path success: {:?}", path)
1228 TestResult::ProbeFailure { path } => {
1229 panic!("Unexpected probe failure: {:?}", path)
1231 TestResult::ProbeSuccess { path } => {
1232 assert_eq!(actual_path, &path);
1237 fn time_passed(&mut self, _: Duration) {}
1241 impl lightning::routing::scoring::Score for TestScorer {}
1243 impl Drop for TestScorer {
1244 fn drop(&mut self) {
1245 if std::thread::panicking() {
1249 if let Some(event_expectations) = &self.event_expectations {
1250 if !event_expectations.is_empty() {
1251 panic!("Unsatisfied event expectations: {:?}", event_expectations);
1257 struct TestWallet {}
1259 impl ChangeDestinationSource for TestWallet {
1260 fn get_change_destination_script(&self) -> Result<ScriptBuf, ()> {
1261 Ok(ScriptBuf::new())
1265 fn get_full_filepath(filepath: String, filename: String) -> String {
1266 let mut path = PathBuf::from(filepath);
1267 path.push(filename);
1268 path.to_str().unwrap().to_string()
1271 fn create_nodes(num_nodes: usize, persist_dir: &str) -> (String, Vec<Node>) {
1272 let persist_temp_path = env::temp_dir().join(persist_dir);
1273 let persist_dir = persist_temp_path.to_string_lossy().to_string();
1274 let network = Network::Bitcoin;
1275 let mut nodes = Vec::new();
1276 for i in 0..num_nodes {
1277 let tx_broadcaster = Arc::new(test_utils::TestBroadcaster::new(network));
1278 let fee_estimator = Arc::new(test_utils::TestFeeEstimator { sat_per_kw: Mutex::new(253) });
1279 let logger = Arc::new(test_utils::TestLogger::with_id(format!("node {}", i)));
1280 let genesis_block = genesis_block(network);
1281 let network_graph = Arc::new(NetworkGraph::new(network, logger.clone()));
1282 let scorer = Arc::new(LockingWrapper::new(TestScorer::new()));
1283 let now = Duration::from_secs(genesis_block.header.time as u64);
1284 let seed = [i as u8; 32];
1285 let keys_manager = Arc::new(KeysManager::new(&seed, now.as_secs(), now.subsec_nanos()));
1286 let router = Arc::new(DefaultRouter::new(network_graph.clone(), logger.clone(), Arc::clone(&keys_manager), scorer.clone(), Default::default()));
1287 let chain_source = Arc::new(test_utils::TestChainSource::new(Network::Bitcoin));
1288 let kv_store = Arc::new(FilesystemStore::new(format!("{}_persister_{}", &persist_dir, i).into()));
1289 let now = Duration::from_secs(genesis_block.header.time as u64);
1290 let keys_manager = Arc::new(KeysManager::new(&seed, now.as_secs(), now.subsec_nanos()));
1291 let chain_monitor = Arc::new(chainmonitor::ChainMonitor::new(Some(chain_source.clone()), tx_broadcaster.clone(), logger.clone(), fee_estimator.clone(), kv_store.clone()));
1292 let best_block = BestBlock::from_network(network);
1293 let params = ChainParameters { network, best_block };
1294 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));
1295 let wallet = Arc::new(TestWallet {});
1296 let sweeper = Arc::new(OutputSweeper::new(best_block, Arc::clone(&tx_broadcaster), Arc::clone(&fee_estimator),
1297 None::<Arc<dyn Filter + Sync + Send>>, Arc::clone(&keys_manager), wallet, Arc::clone(&kv_store), Arc::clone(&logger)));
1298 let p2p_gossip_sync = Arc::new(P2PGossipSync::new(network_graph.clone(), Some(chain_source.clone()), logger.clone()));
1299 let rapid_gossip_sync = Arc::new(RapidGossipSync::new(network_graph.clone(), logger.clone()));
1300 let msg_handler = MessageHandler {
1301 chan_handler: Arc::new(test_utils::TestChannelMessageHandler::new(ChainHash::using_genesis_block(Network::Testnet))),
1302 route_handler: Arc::new(test_utils::TestRoutingMessageHandler::new()),
1303 onion_message_handler: IgnoringMessageHandler{}, custom_message_handler: IgnoringMessageHandler{}
1305 let peer_manager = Arc::new(PeerManager::new(msg_handler, 0, &seed, logger.clone(), keys_manager.clone()));
1306 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, sweeper };
1310 for i in 0..num_nodes {
1311 for j in (i+1)..num_nodes {
1312 nodes[i].node.peer_connected(&nodes[j].node.get_our_node_id(), &Init {
1313 features: nodes[j].node.init_features(), networks: None, remote_network_address: None
1315 nodes[j].node.peer_connected(&nodes[i].node.get_our_node_id(), &Init {
1316 features: nodes[i].node.init_features(), networks: None, remote_network_address: None
1321 (persist_dir, nodes)
1324 macro_rules! open_channel {
1325 ($node_a: expr, $node_b: expr, $channel_value: expr) => {{
1326 begin_open_channel!($node_a, $node_b, $channel_value);
1327 let events = $node_a.node.get_and_clear_pending_events();
1328 assert_eq!(events.len(), 1);
1329 let (temporary_channel_id, tx) = handle_funding_generation_ready!(events[0], $channel_value);
1330 $node_a.node.funding_transaction_generated(&temporary_channel_id, &$node_b.node.get_our_node_id(), tx.clone()).unwrap();
1331 $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()));
1332 get_event!($node_b, Event::ChannelPending);
1333 $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()));
1334 get_event!($node_a, Event::ChannelPending);
1339 macro_rules! begin_open_channel {
1340 ($node_a: expr, $node_b: expr, $channel_value: expr) => {{
1341 $node_a.node.create_channel($node_b.node.get_our_node_id(), $channel_value, 100, 42, None, None).unwrap();
1342 $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()));
1343 $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()));
1347 macro_rules! handle_funding_generation_ready {
1348 ($event: expr, $channel_value: expr) => {{
1350 Event::FundingGenerationReady { temporary_channel_id, channel_value_satoshis, ref output_script, user_channel_id, .. } => {
1351 assert_eq!(channel_value_satoshis, $channel_value);
1352 assert_eq!(user_channel_id, 42);
1354 let tx = Transaction { version: Version::ONE, lock_time: LockTime::ZERO, input: Vec::new(), output: vec![TxOut {
1355 value: Amount::from_sat(channel_value_satoshis), script_pubkey: output_script.clone(),
1357 (temporary_channel_id, tx)
1359 _ => panic!("Unexpected event"),
1364 fn confirm_transaction_depth(node: &mut Node, tx: &Transaction, depth: u32) {
1365 for i in 1..=depth {
1366 let prev_blockhash = node.best_block.block_hash;
1367 let height = node.best_block.height + 1;
1368 let header = create_dummy_header(prev_blockhash, height);
1369 let txdata = vec![(0, tx)];
1370 node.best_block = BestBlock::new(header.block_hash(), height);
1373 node.node.transactions_confirmed(&header, &txdata, height);
1374 node.chain_monitor.transactions_confirmed(&header, &txdata, height);
1375 node.sweeper.transactions_confirmed(&header, &txdata, height);
1377 x if x == depth => {
1378 // We need the TestBroadcaster to know about the new height so that it doesn't think
1379 // we're violating the time lock requirements of transactions broadcasted at that
1381 node.tx_broadcaster.blocks.lock().unwrap().push((genesis_block(Network::Bitcoin), height));
1382 node.node.best_block_updated(&header, height);
1383 node.chain_monitor.best_block_updated(&header, height);
1384 node.sweeper.best_block_updated(&header, height);
1391 fn advance_chain(node: &mut Node, num_blocks: u32) {
1392 for i in 1..=num_blocks {
1393 let prev_blockhash = node.best_block.block_hash;
1394 let height = node.best_block.height + 1;
1395 let header = create_dummy_header(prev_blockhash, height);
1396 node.best_block = BestBlock::new(header.block_hash(), height);
1397 if i == num_blocks {
1398 // We need the TestBroadcaster to know about the new height so that it doesn't think
1399 // we're violating the time lock requirements of transactions broadcasted at that
1401 node.tx_broadcaster.blocks.lock().unwrap().push((genesis_block(Network::Bitcoin), height));
1402 node.node.best_block_updated(&header, height);
1403 node.chain_monitor.best_block_updated(&header, height);
1404 node.sweeper.best_block_updated(&header, height);
1409 fn confirm_transaction(node: &mut Node, tx: &Transaction) {
1410 confirm_transaction_depth(node, tx, ANTI_REORG_DELAY);
1414 fn test_background_processor() {
1415 // Test that when a new channel is created, the ChannelManager needs to be re-persisted with
1416 // updates. Also test that when new updates are available, the manager signals that it needs
1417 // re-persistence and is successfully re-persisted.
1418 let (persist_dir, nodes) = create_nodes(2, "test_background_processor");
1420 // Go through the channel creation process so that each node has something to persist. Since
1421 // open_channel consumes events, it must complete before starting BackgroundProcessor to
1422 // avoid a race with processing events.
1423 let tx = open_channel!(nodes[0], nodes[1], 100000);
1425 // Initiate the background processors to watch each node.
1426 let data_dir = nodes[0].kv_store.get_data_dir();
1427 let persister = Arc::new(Persister::new(data_dir));
1428 let event_handler = |_: _| {};
1429 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()));
1431 macro_rules! check_persisted_data {
1432 ($node: expr, $filepath: expr) => {
1433 let mut expected_bytes = Vec::new();
1435 expected_bytes.clear();
1436 match $node.write(&mut expected_bytes) {
1438 match std::fs::read($filepath) {
1440 if bytes == expected_bytes {
1449 Err(e) => panic!("Unexpected error: {}", e)
1455 // Check that the initial channel manager data is persisted as expected.
1456 let filepath = get_full_filepath(format!("{}_persister_0", &persist_dir), "manager".to_string());
1457 check_persisted_data!(nodes[0].node, filepath.clone());
1460 if !nodes[0].node.get_event_or_persist_condvar_value() { break }
1463 // Force-close the channel.
1464 nodes[0].node.force_close_broadcasting_latest_txn(&ChannelId::v1_from_funding_outpoint(OutPoint { txid: tx.txid(), index: 0 }), &nodes[1].node.get_our_node_id()).unwrap();
1466 // Check that the force-close updates are persisted.
1467 check_persisted_data!(nodes[0].node, filepath.clone());
1469 if !nodes[0].node.get_event_or_persist_condvar_value() { break }
1472 // Check network graph is persisted
1473 let filepath = get_full_filepath(format!("{}_persister_0", &persist_dir), "network_graph".to_string());
1474 check_persisted_data!(nodes[0].network_graph, filepath.clone());
1476 // Check scorer is persisted
1477 let filepath = get_full_filepath(format!("{}_persister_0", &persist_dir), "scorer".to_string());
1478 check_persisted_data!(nodes[0].scorer, filepath.clone());
1480 if !std::thread::panicking() {
1481 bg_processor.stop().unwrap();
1486 fn test_timer_tick_called() {
1488 // - `ChannelManager::timer_tick_occurred` is called every `FRESHNESS_TIMER`,
1489 // - `ChainMonitor::rebroadcast_pending_claims` is called every `REBROADCAST_TIMER`,
1490 // - `PeerManager::timer_tick_occurred` is called every `PING_TIMER`, and
1491 // - `OnionMessageHandler::timer_tick_occurred` is called every `ONION_MESSAGE_HANDLER_TIMER`.
1492 let (_, nodes) = create_nodes(1, "test_timer_tick_called");
1493 let data_dir = nodes[0].kv_store.get_data_dir();
1494 let persister = Arc::new(Persister::new(data_dir));
1495 let event_handler = |_: _| {};
1496 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()));
1498 let log_entries = nodes[0].logger.lines.lock().unwrap();
1499 let desired_log_1 = "Calling ChannelManager's timer_tick_occurred".to_string();
1500 let desired_log_2 = "Calling PeerManager's timer_tick_occurred".to_string();
1501 let desired_log_3 = "Rebroadcasting monitor's pending claims".to_string();
1502 let desired_log_4 = "Calling OnionMessageHandler's timer_tick_occurred".to_string();
1503 if log_entries.get(&("lightning_background_processor", desired_log_1)).is_some() &&
1504 log_entries.get(&("lightning_background_processor", desired_log_2)).is_some() &&
1505 log_entries.get(&("lightning_background_processor", desired_log_3)).is_some() &&
1506 log_entries.get(&("lightning_background_processor", desired_log_4)).is_some() {
1511 if !std::thread::panicking() {
1512 bg_processor.stop().unwrap();
1517 fn test_channel_manager_persist_error() {
1518 // Test that if we encounter an error during manager persistence, the thread panics.
1519 let (_, nodes) = create_nodes(2, "test_persist_error");
1520 open_channel!(nodes[0], nodes[1], 100000);
1522 let data_dir = nodes[0].kv_store.get_data_dir();
1523 let persister = Arc::new(Persister::new(data_dir).with_manager_error(std::io::ErrorKind::Other, "test"));
1524 let event_handler = |_: _| {};
1525 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()));
1526 match bg_processor.join() {
1527 Ok(_) => panic!("Expected error persisting manager"),
1529 assert_eq!(e.kind(), std::io::ErrorKind::Other);
1530 assert_eq!(e.get_ref().unwrap().to_string(), "test");
1536 #[cfg(feature = "futures")]
1537 async fn test_channel_manager_persist_error_async() {
1538 // Test that if we encounter an error during manager persistence, the thread panics.
1539 let (_, nodes) = create_nodes(2, "test_persist_error_sync");
1540 open_channel!(nodes[0], nodes[1], 100000);
1542 let data_dir = nodes[0].kv_store.get_data_dir();
1543 let persister = Arc::new(Persister::new(data_dir).with_manager_error(std::io::ErrorKind::Other, "test"));
1545 let bp_future = super::process_events_async(
1546 persister, |_: _| {async {}}, nodes[0].chain_monitor.clone(), nodes[0].node.clone(),
1547 nodes[0].rapid_gossip_sync(), nodes[0].peer_manager.clone(), nodes[0].logger.clone(),
1548 Some(nodes[0].scorer.clone()), move |dur: Duration| {
1549 Box::pin(async move {
1550 tokio::time::sleep(dur).await;
1553 }, false, || Some(Duration::ZERO),
1555 match bp_future.await {
1556 Ok(_) => panic!("Expected error persisting manager"),
1558 assert_eq!(e.kind(), std::io::ErrorKind::Other);
1559 assert_eq!(e.get_ref().unwrap().to_string(), "test");
1565 fn test_network_graph_persist_error() {
1566 // Test that if we encounter an error during network graph persistence, an error gets returned.
1567 let (_, nodes) = create_nodes(2, "test_persist_network_graph_error");
1568 let data_dir = nodes[0].kv_store.get_data_dir();
1569 let persister = Arc::new(Persister::new(data_dir).with_graph_error(std::io::ErrorKind::Other, "test"));
1570 let event_handler = |_: _| {};
1571 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()));
1573 match bg_processor.stop() {
1574 Ok(_) => panic!("Expected error persisting network graph"),
1576 assert_eq!(e.kind(), std::io::ErrorKind::Other);
1577 assert_eq!(e.get_ref().unwrap().to_string(), "test");
1583 fn test_scorer_persist_error() {
1584 // Test that if we encounter an error during scorer persistence, an error gets returned.
1585 let (_, nodes) = create_nodes(2, "test_persist_scorer_error");
1586 let data_dir = nodes[0].kv_store.get_data_dir();
1587 let persister = Arc::new(Persister::new(data_dir).with_scorer_error(std::io::ErrorKind::Other, "test"));
1588 let event_handler = |_: _| {};
1589 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()));
1591 match bg_processor.stop() {
1592 Ok(_) => panic!("Expected error persisting scorer"),
1594 assert_eq!(e.kind(), std::io::ErrorKind::Other);
1595 assert_eq!(e.get_ref().unwrap().to_string(), "test");
1601 fn test_background_event_handling() {
1602 let (_, mut nodes) = create_nodes(2, "test_background_event_handling");
1603 let channel_value = 100000;
1604 let data_dir = nodes[0].kv_store.get_data_dir();
1605 let persister = Arc::new(Persister::new(data_dir.clone()));
1607 // Set up a background event handler for FundingGenerationReady events.
1608 let (funding_generation_send, funding_generation_recv) = std::sync::mpsc::sync_channel(1);
1609 let (channel_pending_send, channel_pending_recv) = std::sync::mpsc::sync_channel(1);
1610 let event_handler = move |event: Event| match event {
1611 Event::FundingGenerationReady { .. } => funding_generation_send.send(handle_funding_generation_ready!(event, channel_value)).unwrap(),
1612 Event::ChannelPending { .. } => channel_pending_send.send(()).unwrap(),
1613 Event::ChannelReady { .. } => {},
1614 _ => panic!("Unexpected event: {:?}", event),
1617 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()));
1619 // Open a channel and check that the FundingGenerationReady event was handled.
1620 begin_open_channel!(nodes[0], nodes[1], channel_value);
1621 let (temporary_channel_id, funding_tx) = funding_generation_recv
1622 .recv_timeout(Duration::from_secs(EVENT_DEADLINE))
1623 .expect("FundingGenerationReady not handled within deadline");
1624 nodes[0].node.funding_transaction_generated(&temporary_channel_id, &nodes[1].node.get_our_node_id(), funding_tx.clone()).unwrap();
1625 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()));
1626 get_event!(nodes[1], Event::ChannelPending);
1627 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()));
1628 let _ = channel_pending_recv.recv_timeout(Duration::from_secs(EVENT_DEADLINE))
1629 .expect("ChannelPending not handled within deadline");
1631 // Confirm the funding transaction.
1632 confirm_transaction(&mut nodes[0], &funding_tx);
1633 let as_funding = get_event_msg!(nodes[0], MessageSendEvent::SendChannelReady, nodes[1].node.get_our_node_id());
1634 confirm_transaction(&mut nodes[1], &funding_tx);
1635 let bs_funding = get_event_msg!(nodes[1], MessageSendEvent::SendChannelReady, nodes[0].node.get_our_node_id());
1636 nodes[0].node.handle_channel_ready(&nodes[1].node.get_our_node_id(), &bs_funding);
1637 let _as_channel_update = get_event_msg!(nodes[0], MessageSendEvent::SendChannelUpdate, nodes[1].node.get_our_node_id());
1638 nodes[1].node.handle_channel_ready(&nodes[0].node.get_our_node_id(), &as_funding);
1639 let _bs_channel_update = get_event_msg!(nodes[1], MessageSendEvent::SendChannelUpdate, nodes[0].node.get_our_node_id());
1640 let broadcast_funding = nodes[0].tx_broadcaster.txn_broadcasted.lock().unwrap().pop().unwrap();
1641 assert_eq!(broadcast_funding.txid(), funding_tx.txid());
1642 assert!(nodes[0].tx_broadcaster.txn_broadcasted.lock().unwrap().is_empty());
1644 if !std::thread::panicking() {
1645 bg_processor.stop().unwrap();
1648 // Set up a background event handler for SpendableOutputs events.
1649 let (sender, receiver) = std::sync::mpsc::sync_channel(1);
1650 let event_handler = move |event: Event| match event {
1651 Event::SpendableOutputs { .. } => sender.send(event).unwrap(),
1652 Event::ChannelReady { .. } => {},
1653 Event::ChannelClosed { .. } => {},
1654 _ => panic!("Unexpected event: {:?}", event),
1656 let persister = Arc::new(Persister::new(data_dir));
1657 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()));
1659 // Force close the channel and check that the SpendableOutputs event was handled.
1660 nodes[0].node.force_close_broadcasting_latest_txn(&nodes[0].node.list_channels()[0].channel_id, &nodes[1].node.get_our_node_id()).unwrap();
1661 let commitment_tx = nodes[0].tx_broadcaster.txn_broadcasted.lock().unwrap().pop().unwrap();
1662 confirm_transaction_depth(&mut nodes[0], &commitment_tx, BREAKDOWN_TIMEOUT as u32);
1664 let event = receiver
1665 .recv_timeout(Duration::from_secs(EVENT_DEADLINE))
1666 .expect("Events not handled within deadline");
1668 Event::SpendableOutputs { outputs, channel_id } => {
1669 nodes[0].sweeper.track_spendable_outputs(outputs, channel_id, false, Some(153)).unwrap();
1671 _ => panic!("Unexpected event: {:?}", event),
1674 // Check we don't generate an initial sweeping tx until we reach the required height.
1675 assert_eq!(nodes[0].sweeper.tracked_spendable_outputs().len(), 1);
1676 let tracked_output = nodes[0].sweeper.tracked_spendable_outputs().first().unwrap().clone();
1677 if let Some(sweep_tx_0) = nodes[0].tx_broadcaster.txn_broadcasted.lock().unwrap().pop() {
1678 assert!(!tracked_output.is_spent_in(&sweep_tx_0));
1679 match tracked_output.status {
1680 OutputSpendStatus::PendingInitialBroadcast { delayed_until_height } => {
1681 assert_eq!(delayed_until_height, Some(153));
1683 _ => panic!("Unexpected status"),
1687 advance_chain(&mut nodes[0], 3);
1689 // Check we generate an initial sweeping tx.
1690 assert_eq!(nodes[0].sweeper.tracked_spendable_outputs().len(), 1);
1691 let tracked_output = nodes[0].sweeper.tracked_spendable_outputs().first().unwrap().clone();
1692 let sweep_tx_0 = nodes[0].tx_broadcaster.txn_broadcasted.lock().unwrap().pop().unwrap();
1693 match tracked_output.status {
1694 OutputSpendStatus::PendingFirstConfirmation { latest_spending_tx, .. } => {
1695 assert_eq!(sweep_tx_0.txid(), latest_spending_tx.txid());
1697 _ => panic!("Unexpected status"),
1700 // Check we regenerate and rebroadcast the sweeping tx each block.
1701 advance_chain(&mut nodes[0], 1);
1702 assert_eq!(nodes[0].sweeper.tracked_spendable_outputs().len(), 1);
1703 let tracked_output = nodes[0].sweeper.tracked_spendable_outputs().first().unwrap().clone();
1704 let sweep_tx_1 = nodes[0].tx_broadcaster.txn_broadcasted.lock().unwrap().pop().unwrap();
1705 match tracked_output.status {
1706 OutputSpendStatus::PendingFirstConfirmation { latest_spending_tx, .. } => {
1707 assert_eq!(sweep_tx_1.txid(), latest_spending_tx.txid());
1709 _ => panic!("Unexpected status"),
1711 assert_ne!(sweep_tx_0, sweep_tx_1);
1713 advance_chain(&mut nodes[0], 1);
1714 assert_eq!(nodes[0].sweeper.tracked_spendable_outputs().len(), 1);
1715 let tracked_output = nodes[0].sweeper.tracked_spendable_outputs().first().unwrap().clone();
1716 let sweep_tx_2 = nodes[0].tx_broadcaster.txn_broadcasted.lock().unwrap().pop().unwrap();
1717 match tracked_output.status {
1718 OutputSpendStatus::PendingFirstConfirmation { latest_spending_tx, .. } => {
1719 assert_eq!(sweep_tx_2.txid(), latest_spending_tx.txid());
1721 _ => panic!("Unexpected status"),
1723 assert_ne!(sweep_tx_0, sweep_tx_2);
1724 assert_ne!(sweep_tx_1, sweep_tx_2);
1726 // Check we still track the spendable outputs up to ANTI_REORG_DELAY confirmations.
1727 confirm_transaction_depth(&mut nodes[0], &sweep_tx_2, 5);
1728 assert_eq!(nodes[0].sweeper.tracked_spendable_outputs().len(), 1);
1729 let tracked_output = nodes[0].sweeper.tracked_spendable_outputs().first().unwrap().clone();
1730 match tracked_output.status {
1731 OutputSpendStatus::PendingThresholdConfirmations { latest_spending_tx, .. } => {
1732 assert_eq!(sweep_tx_2.txid(), latest_spending_tx.txid());
1734 _ => panic!("Unexpected status"),
1737 // Check we still see the transaction as confirmed if we unconfirm any untracked
1738 // transaction. (We previously had a bug that would mark tracked transactions as
1739 // unconfirmed if any transaction at an unknown block height would be unconfirmed.)
1740 let unconf_txid = Txid::from_slice(&[0; 32]).unwrap();
1741 nodes[0].sweeper.transaction_unconfirmed(&unconf_txid);
1743 assert_eq!(nodes[0].sweeper.tracked_spendable_outputs().len(), 1);
1744 let tracked_output = nodes[0].sweeper.tracked_spendable_outputs().first().unwrap().clone();
1745 match tracked_output.status {
1746 OutputSpendStatus::PendingThresholdConfirmations { latest_spending_tx, .. } => {
1747 assert_eq!(sweep_tx_2.txid(), latest_spending_tx.txid());
1749 _ => panic!("Unexpected status"),
1752 // Check we stop tracking the spendable outputs when one of the txs reaches
1753 // ANTI_REORG_DELAY confirmations.
1754 confirm_transaction_depth(&mut nodes[0], &sweep_tx_0, ANTI_REORG_DELAY);
1755 assert_eq!(nodes[0].sweeper.tracked_spendable_outputs().len(), 0);
1757 if !std::thread::panicking() {
1758 bg_processor.stop().unwrap();
1763 fn test_scorer_persistence() {
1764 let (_, nodes) = create_nodes(2, "test_scorer_persistence");
1765 let data_dir = nodes[0].kv_store.get_data_dir();
1766 let persister = Arc::new(Persister::new(data_dir));
1767 let event_handler = |_: _| {};
1768 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()));
1771 let log_entries = nodes[0].logger.lines.lock().unwrap();
1772 let expected_log = "Calling time_passed and persisting scorer".to_string();
1773 if log_entries.get(&("lightning_background_processor", expected_log)).is_some() {
1778 if !std::thread::panicking() {
1779 bg_processor.stop().unwrap();
1783 macro_rules! do_test_not_pruning_network_graph_until_graph_sync_completion {
1784 ($nodes: expr, $receive: expr, $sleep: expr) => {
1785 let features = ChannelFeatures::empty();
1786 $nodes[0].network_graph.add_channel_from_partial_announcement(
1787 42, 53, features, $nodes[0].node.get_our_node_id(), $nodes[1].node.get_our_node_id()
1788 ).expect("Failed to update channel from partial announcement");
1789 let original_graph_description = $nodes[0].network_graph.to_string();
1790 assert!(original_graph_description.contains("42: features: 0000, node_one:"));
1791 assert_eq!($nodes[0].network_graph.read_only().channels().len(), 1);
1795 let log_entries = $nodes[0].logger.lines.lock().unwrap();
1796 let loop_counter = "Calling ChannelManager's timer_tick_occurred".to_string();
1797 if *log_entries.get(&("lightning_background_processor", loop_counter))
1800 // Wait until the loop has gone around at least twice.
1805 let initialization_input = vec![
1806 76, 68, 75, 1, 111, 226, 140, 10, 182, 241, 179, 114, 193, 166, 162, 70, 174, 99, 247,
1807 79, 147, 30, 131, 101, 225, 90, 8, 156, 104, 214, 25, 0, 0, 0, 0, 0, 97, 227, 98, 218,
1808 0, 0, 0, 4, 2, 22, 7, 207, 206, 25, 164, 197, 231, 230, 231, 56, 102, 61, 250, 251,
1809 187, 172, 38, 46, 79, 247, 108, 44, 155, 48, 219, 238, 252, 53, 192, 6, 67, 2, 36, 125,
1810 157, 176, 223, 175, 234, 116, 94, 248, 201, 225, 97, 235, 50, 47, 115, 172, 63, 136,
1811 88, 216, 115, 11, 111, 217, 114, 84, 116, 124, 231, 107, 2, 158, 1, 242, 121, 152, 106,
1812 204, 131, 186, 35, 93, 70, 216, 10, 237, 224, 183, 89, 95, 65, 3, 83, 185, 58, 138,
1813 181, 64, 187, 103, 127, 68, 50, 2, 201, 19, 17, 138, 136, 149, 185, 226, 156, 137, 175,
1814 110, 32, 237, 0, 217, 90, 31, 100, 228, 149, 46, 219, 175, 168, 77, 4, 143, 38, 128,
1815 76, 97, 0, 0, 0, 2, 0, 0, 255, 8, 153, 192, 0, 2, 27, 0, 0, 0, 1, 0, 0, 255, 2, 68,
1816 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,
1817 0, 0, 0, 1, 0, 0, 0, 0, 58, 85, 116, 216, 255, 8, 153, 192, 0, 2, 27, 0, 0, 25, 0, 0,
1818 0, 1, 0, 0, 0, 125, 255, 2, 68, 226, 0, 6, 11, 0, 1, 5, 0, 0, 0, 0, 29, 129, 25, 192,
1820 $nodes[0].rapid_gossip_sync.update_network_graph_no_std(&initialization_input[..], Some(1642291930)).unwrap();
1822 // this should have added two channels and pruned the previous one.
1823 assert_eq!($nodes[0].network_graph.read_only().channels().len(), 2);
1825 $receive.expect("Network graph not pruned within deadline");
1827 // all channels should now be pruned
1828 assert_eq!($nodes[0].network_graph.read_only().channels().len(), 0);
1833 fn test_not_pruning_network_graph_until_graph_sync_completion() {
1834 let (sender, receiver) = std::sync::mpsc::sync_channel(1);
1836 let (_, nodes) = create_nodes(2, "test_not_pruning_network_graph_until_graph_sync_completion");
1837 let data_dir = nodes[0].kv_store.get_data_dir();
1838 let persister = Arc::new(Persister::new(data_dir).with_graph_persistence_notifier(sender));
1840 let event_handler = |_: _| {};
1841 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()));
1843 do_test_not_pruning_network_graph_until_graph_sync_completion!(nodes,
1844 receiver.recv_timeout(Duration::from_secs(super::FIRST_NETWORK_PRUNE_TIMER * 5)),
1845 std::thread::sleep(Duration::from_millis(1)));
1847 background_processor.stop().unwrap();
1851 #[cfg(feature = "futures")]
1852 async fn test_not_pruning_network_graph_until_graph_sync_completion_async() {
1853 let (sender, receiver) = std::sync::mpsc::sync_channel(1);
1855 let (_, nodes) = create_nodes(2, "test_not_pruning_network_graph_until_graph_sync_completion_async");
1856 let data_dir = nodes[0].kv_store.get_data_dir();
1857 let persister = Arc::new(Persister::new(data_dir).with_graph_persistence_notifier(sender));
1859 let (exit_sender, exit_receiver) = tokio::sync::watch::channel(());
1860 let bp_future = super::process_events_async(
1861 persister, |_: _| {async {}}, nodes[0].chain_monitor.clone(), nodes[0].node.clone(),
1862 nodes[0].rapid_gossip_sync(), nodes[0].peer_manager.clone(), nodes[0].logger.clone(),
1863 Some(nodes[0].scorer.clone()), move |dur: Duration| {
1864 let mut exit_receiver = exit_receiver.clone();
1865 Box::pin(async move {
1867 _ = tokio::time::sleep(dur) => false,
1868 _ = exit_receiver.changed() => true,
1871 }, false, || Some(Duration::from_secs(1696300000)),
1874 let t1 = tokio::spawn(bp_future);
1875 let t2 = tokio::spawn(async move {
1876 do_test_not_pruning_network_graph_until_graph_sync_completion!(nodes, {
1879 tokio::time::sleep(Duration::from_secs(super::FIRST_NETWORK_PRUNE_TIMER)).await;
1880 if let Ok(()) = receiver.try_recv() { break Ok::<(), ()>(()); }
1884 }, tokio::time::sleep(Duration::from_millis(1)).await);
1885 exit_sender.send(()).unwrap();
1887 let (r1, r2) = tokio::join!(t1, t2);
1888 r1.unwrap().unwrap();
1892 macro_rules! do_test_payment_path_scoring {
1893 ($nodes: expr, $receive: expr) => {
1894 // Ensure that we update the scorer when relevant events are processed. In this case, we ensure
1895 // that we update the scorer upon a payment path succeeding (note that the channel must be
1896 // public or else we won't score it).
1897 // A background event handler for FundingGenerationReady events must be hooked up to a
1898 // running background processor.
1899 let scored_scid = 4242;
1900 let secp_ctx = Secp256k1::new();
1901 let node_1_privkey = SecretKey::from_slice(&[42; 32]).unwrap();
1902 let node_1_id = PublicKey::from_secret_key(&secp_ctx, &node_1_privkey);
1904 let path = Path { hops: vec![RouteHop {
1906 node_features: NodeFeatures::empty(),
1907 short_channel_id: scored_scid,
1908 channel_features: ChannelFeatures::empty(),
1910 cltv_expiry_delta: MIN_CLTV_EXPIRY_DELTA as u32,
1911 maybe_announced_channel: true,
1912 }], blinded_tail: None };
1914 $nodes[0].scorer.write_lock().expect(TestResult::PaymentFailure { path: path.clone(), short_channel_id: scored_scid });
1915 $nodes[0].node.push_pending_event(Event::PaymentPathFailed {
1917 payment_hash: PaymentHash([42; 32]),
1918 payment_failed_permanently: false,
1919 failure: PathFailure::OnPath { network_update: None },
1921 short_channel_id: Some(scored_scid),
1923 let event = $receive.expect("PaymentPathFailed not handled within deadline");
1925 Event::PaymentPathFailed { .. } => {},
1926 _ => panic!("Unexpected event"),
1929 // Ensure we'll score payments that were explicitly failed back by the destination as
1931 $nodes[0].scorer.write_lock().expect(TestResult::ProbeSuccess { path: path.clone() });
1932 $nodes[0].node.push_pending_event(Event::PaymentPathFailed {
1934 payment_hash: PaymentHash([42; 32]),
1935 payment_failed_permanently: true,
1936 failure: PathFailure::OnPath { network_update: None },
1938 short_channel_id: None,
1940 let event = $receive.expect("PaymentPathFailed not handled within deadline");
1942 Event::PaymentPathFailed { .. } => {},
1943 _ => panic!("Unexpected event"),
1946 $nodes[0].scorer.write_lock().expect(TestResult::PaymentSuccess { path: path.clone() });
1947 $nodes[0].node.push_pending_event(Event::PaymentPathSuccessful {
1948 payment_id: PaymentId([42; 32]),
1952 let event = $receive.expect("PaymentPathSuccessful not handled within deadline");
1954 Event::PaymentPathSuccessful { .. } => {},
1955 _ => panic!("Unexpected event"),
1958 $nodes[0].scorer.write_lock().expect(TestResult::ProbeSuccess { path: path.clone() });
1959 $nodes[0].node.push_pending_event(Event::ProbeSuccessful {
1960 payment_id: PaymentId([42; 32]),
1961 payment_hash: PaymentHash([42; 32]),
1964 let event = $receive.expect("ProbeSuccessful not handled within deadline");
1966 Event::ProbeSuccessful { .. } => {},
1967 _ => panic!("Unexpected event"),
1970 $nodes[0].scorer.write_lock().expect(TestResult::ProbeFailure { path: path.clone() });
1971 $nodes[0].node.push_pending_event(Event::ProbeFailed {
1972 payment_id: PaymentId([42; 32]),
1973 payment_hash: PaymentHash([42; 32]),
1975 short_channel_id: Some(scored_scid),
1977 let event = $receive.expect("ProbeFailure not handled within deadline");
1979 Event::ProbeFailed { .. } => {},
1980 _ => panic!("Unexpected event"),
1986 fn test_payment_path_scoring() {
1987 let (sender, receiver) = std::sync::mpsc::sync_channel(1);
1988 let event_handler = move |event: Event| match event {
1989 Event::PaymentPathFailed { .. } => sender.send(event).unwrap(),
1990 Event::PaymentPathSuccessful { .. } => sender.send(event).unwrap(),
1991 Event::ProbeSuccessful { .. } => sender.send(event).unwrap(),
1992 Event::ProbeFailed { .. } => sender.send(event).unwrap(),
1993 _ => panic!("Unexpected event: {:?}", event),
1996 let (_, nodes) = create_nodes(1, "test_payment_path_scoring");
1997 let data_dir = nodes[0].kv_store.get_data_dir();
1998 let persister = Arc::new(Persister::new(data_dir));
1999 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()));
2001 do_test_payment_path_scoring!(nodes, receiver.recv_timeout(Duration::from_secs(EVENT_DEADLINE)));
2003 if !std::thread::panicking() {
2004 bg_processor.stop().unwrap();
2007 let log_entries = nodes[0].logger.lines.lock().unwrap();
2008 let expected_log = "Persisting scorer after update".to_string();
2009 assert_eq!(*log_entries.get(&("lightning_background_processor", expected_log)).unwrap(), 5);
2013 #[cfg(feature = "futures")]
2014 async fn test_payment_path_scoring_async() {
2015 let (sender, mut receiver) = tokio::sync::mpsc::channel(1);
2016 let event_handler = move |event: Event| {
2017 let sender_ref = sender.clone();
2020 Event::PaymentPathFailed { .. } => { sender_ref.send(event).await.unwrap() },
2021 Event::PaymentPathSuccessful { .. } => { sender_ref.send(event).await.unwrap() },
2022 Event::ProbeSuccessful { .. } => { sender_ref.send(event).await.unwrap() },
2023 Event::ProbeFailed { .. } => { sender_ref.send(event).await.unwrap() },
2024 _ => panic!("Unexpected event: {:?}", event),
2029 let (_, nodes) = create_nodes(1, "test_payment_path_scoring_async");
2030 let data_dir = nodes[0].kv_store.get_data_dir();
2031 let persister = Arc::new(Persister::new(data_dir));
2033 let (exit_sender, exit_receiver) = tokio::sync::watch::channel(());
2035 let bp_future = super::process_events_async(
2036 persister, event_handler, nodes[0].chain_monitor.clone(), nodes[0].node.clone(),
2037 nodes[0].no_gossip_sync(), nodes[0].peer_manager.clone(), nodes[0].logger.clone(),
2038 Some(nodes[0].scorer.clone()), move |dur: Duration| {
2039 let mut exit_receiver = exit_receiver.clone();
2040 Box::pin(async move {
2042 _ = tokio::time::sleep(dur) => false,
2043 _ = exit_receiver.changed() => true,
2046 }, false, || Some(Duration::ZERO),
2048 let t1 = tokio::spawn(bp_future);
2049 let t2 = tokio::spawn(async move {
2050 do_test_payment_path_scoring!(nodes, receiver.recv().await);
2051 exit_sender.send(()).unwrap();
2053 let log_entries = nodes[0].logger.lines.lock().unwrap();
2054 let expected_log = "Persisting scorer after update".to_string();
2055 assert_eq!(*log_entries.get(&("lightning_background_processor", expected_log)).unwrap(), 5);
2058 let (r1, r2) = tokio::join!(t1, t2);
2059 r1.unwrap().unwrap();