use util::config::{UserConfig, ChannelConfig};
use util::events::{EventHandler, EventsProvider, MessageSendEvent, MessageSendEventsProvider, ClosureReason, HTLCDestination};
use util::{byte_utils, events};
+use util::crypto::sign;
+use util::wakers::{Future, Notifier};
use util::scid_utils::fake_scid;
use util::ser::{BigSize, FixedLengthReader, Readable, ReadableArgs, MaybeReadable, Writeable, Writer, VecWriter};
use util::logger::{Level, Logger};
use core::{cmp, mem};
use core::cell::RefCell;
use io::Read;
-use sync::{Arc, Condvar, Mutex, MutexGuard, RwLock, RwLockReadGuard};
+use sync::{Arc, Mutex, MutexGuard, RwLock, RwLockReadGuard};
use core::sync::atomic::{AtomicUsize, Ordering};
use core::time::Duration;
use core::ops::Deref;
-#[cfg(any(test, feature = "std"))]
-use std::time::Instant;
-use util::crypto::sign;
-
// We hold various information about HTLC relay in the HTLC objects in Channel itself:
//
// Upon receipt of an HTLC from a peer, we'll give it a PendingHTLCStatus indicating if it should
/// Taken first everywhere where we are making changes before any other locks.
/// When acquiring this lock in read mode, rather than acquiring it directly, call
/// `PersistenceNotifierGuard::notify_on_drop(..)` and pass the lock to it, to ensure the
- /// PersistenceNotifier the lock contains sends out a notification when the lock is released.
+ /// Notifier the lock contains sends out a notification when the lock is released.
total_consistency_lock: RwLock<()>,
- persistence_notifier: PersistenceNotifier,
+ persistence_notifier: Notifier,
keys_manager: K,
/// notify or not based on whether relevant changes have been made, providing a closure to
/// `optionally_notify` which returns a `NotifyOption`.
struct PersistenceNotifierGuard<'a, F: Fn() -> NotifyOption> {
- persistence_notifier: &'a PersistenceNotifier,
+ persistence_notifier: &'a Notifier,
should_persist: F,
// We hold onto this result so the lock doesn't get released immediately.
_read_guard: RwLockReadGuard<'a, ()>,
}
impl<'a> PersistenceNotifierGuard<'a, fn() -> NotifyOption> { // We don't care what the concrete F is here, it's unused
- fn notify_on_drop(lock: &'a RwLock<()>, notifier: &'a PersistenceNotifier) -> PersistenceNotifierGuard<'a, impl Fn() -> NotifyOption> {
+ fn notify_on_drop(lock: &'a RwLock<()>, notifier: &'a Notifier) -> PersistenceNotifierGuard<'a, impl Fn() -> NotifyOption> {
PersistenceNotifierGuard::optionally_notify(lock, notifier, || -> NotifyOption { NotifyOption::DoPersist })
}
- fn optionally_notify<F: Fn() -> NotifyOption>(lock: &'a RwLock<()>, notifier: &'a PersistenceNotifier, persist_check: F) -> PersistenceNotifierGuard<'a, F> {
+ fn optionally_notify<F: Fn() -> NotifyOption>(lock: &'a RwLock<()>, notifier: &'a Notifier, persist_check: F) -> PersistenceNotifierGuard<'a, F> {
let read_guard = lock.read().unwrap();
PersistenceNotifierGuard {
pending_events: Mutex::new(Vec::new()),
pending_background_events: Mutex::new(Vec::new()),
total_consistency_lock: RwLock::new(()),
- persistence_notifier: PersistenceNotifier::new(),
+ persistence_notifier: Notifier::new(),
keys_manager,
counterparty_node_id: &PublicKey
) {
for (htlc_src, payment_hash) in htlcs_to_fail.drain(..) {
- match htlc_src {
- HTLCSource::PreviousHopData(HTLCPreviousHopData { .. }) => {
- let (failure_code, onion_failure_data) =
- match self.channel_state.lock().unwrap().by_id.entry(channel_id) {
- hash_map::Entry::Occupied(chan_entry) => {
- self.get_htlc_inbound_temp_fail_err_and_data(0x1000|7, &chan_entry.get())
- },
- hash_map::Entry::Vacant(_) => (0x4000|10, Vec::new())
- };
- let channel_state = self.channel_state.lock().unwrap();
+ let mut channel_state = self.channel_state.lock().unwrap();
+ let (failure_code, onion_failure_data) =
+ match channel_state.by_id.entry(channel_id) {
+ hash_map::Entry::Occupied(chan_entry) => {
+ self.get_htlc_inbound_temp_fail_err_and_data(0x1000|7, &chan_entry.get())
+ },
+ hash_map::Entry::Vacant(_) => (0x4000|10, Vec::new())
+ };
- let receiver = HTLCDestination::NextHopChannel { node_id: Some(counterparty_node_id.clone()), channel_id };
- self.fail_htlc_backwards_internal(channel_state, htlc_src, &payment_hash, HTLCFailReason::Reason { failure_code, data: onion_failure_data }, receiver)
- },
- HTLCSource::OutboundRoute { session_priv, payment_id, path, payment_params, .. } => {
- let mut session_priv_bytes = [0; 32];
- session_priv_bytes.copy_from_slice(&session_priv[..]);
- let mut outbounds = self.pending_outbound_payments.lock().unwrap();
- if let hash_map::Entry::Occupied(mut payment) = outbounds.entry(payment_id) {
- if payment.get_mut().remove(&session_priv_bytes, Some(&path)) && !payment.get().is_fulfilled() {
- let retry = if let Some(payment_params_data) = payment_params {
- let path_last_hop = path.last().expect("Outbound payments must have had a valid path");
- Some(RouteParameters {
- payment_params: payment_params_data,
- final_value_msat: path_last_hop.fee_msat,
- final_cltv_expiry_delta: path_last_hop.cltv_expiry_delta,
- })
- } else { None };
- let mut pending_events = self.pending_events.lock().unwrap();
- pending_events.push(events::Event::PaymentPathFailed {
- payment_id: Some(payment_id),
- payment_hash,
- rejected_by_dest: false,
- network_update: None,
- all_paths_failed: payment.get().remaining_parts() == 0,
- path: path.clone(),
- short_channel_id: None,
- retry,
- #[cfg(test)]
- error_code: None,
- #[cfg(test)]
- error_data: None,
- });
- if payment.get().abandoned() && payment.get().remaining_parts() == 0 {
- pending_events.push(events::Event::PaymentFailed {
- payment_id,
- payment_hash: payment.get().payment_hash().expect("PendingOutboundPayments::RetriesExceeded always has a payment hash set"),
- });
- payment.remove();
- }
- }
- } else {
- log_trace!(self.logger, "Received duplicative fail for HTLC with payment_hash {}", log_bytes!(payment_hash.0));
- }
- },
- };
+ let receiver = HTLCDestination::NextHopChannel { node_id: Some(counterparty_node_id.clone()), channel_id };
+ self.fail_htlc_backwards_internal(channel_state, htlc_src, &payment_hash, HTLCFailReason::Reason { failure_code, data: onion_failure_data }, receiver);
}
}
events::Event::PaymentPathFailed {
payment_id: Some(payment_id),
payment_hash: payment_hash.clone(),
- rejected_by_dest: !payment_retryable,
+ payment_failed_permanently: !payment_retryable,
network_update,
all_paths_failed,
path: path.clone(),
// channel here as we apparently can't relay through them anyway.
let scid = path.first().unwrap().short_channel_id;
retry.as_mut().map(|r| r.payment_params.previously_failed_channels.push(scid));
- events::Event::PaymentPathFailed {
- payment_id: Some(payment_id),
- payment_hash: payment_hash.clone(),
- rejected_by_dest: path.len() == 1,
- network_update: None,
- all_paths_failed,
- path: path.clone(),
- short_channel_id: Some(scid),
- retry,
+
+ if self.payment_is_probe(payment_hash, &payment_id) {
+ events::Event::ProbeFailed {
+ payment_id: payment_id,
+ payment_hash: payment_hash.clone(),
+ path: path.clone(),
+ short_channel_id: Some(scid),
+ }
+ } else {
+ events::Event::PaymentPathFailed {
+ payment_id: Some(payment_id),
+ payment_hash: payment_hash.clone(),
+ payment_failed_permanently: false,
+ network_update: None,
+ all_paths_failed,
+ path: path.clone(),
+ short_channel_id: Some(scid),
+ retry,
#[cfg(test)]
- error_code: Some(*failure_code),
+ error_code: Some(*failure_code),
#[cfg(test)]
- error_data: Some(data.clone()),
+ error_data: Some(data.clone()),
+ }
}
}
};
///
/// An [`EventHandler`] may safely call back to the provider in order to handle an event.
/// However, it must not call [`Writeable::write`] as doing so would result in a deadlock.
- ///
- /// Pending events are persisted as part of [`ChannelManager`]. While these events are cleared
- /// when processed, an [`EventHandler`] must be able to handle previously seen events when
- /// restarting from an old state.
fn process_pending_events<H: Deref>(&self, handler: H) where H::Target: EventHandler {
PersistenceNotifierGuard::optionally_notify(&self.total_consistency_lock, &self.persistence_notifier, || {
let mut result = NotifyOption::SkipPersist;
self.persistence_notifier.wait()
}
+ /// Gets a [`Future`] that completes when a persistable update is available. Note that
+ /// callbacks registered on the [`Future`] MUST NOT call back into this [`ChannelManager`] and
+ /// should instead register actions to be taken later.
+ pub fn get_persistable_update_future(&self) -> Future {
+ self.persistence_notifier.get_future()
+ }
+
#[cfg(any(test, feature = "_test_utils"))]
pub fn get_persistence_condvar_value(&self) -> bool {
- let mutcond = &self.persistence_notifier.persistence_lock;
- let &(ref mtx, _) = mutcond;
- let guard = mtx.lock().unwrap();
- *guard
+ self.persistence_notifier.notify_pending()
}
/// Gets the latest best block which was connected either via the [`chain::Listen`] or
}
}
-/// Used to signal to the ChannelManager persister that the manager needs to be re-persisted to
-/// disk/backups, through `await_persistable_update_timeout` and `await_persistable_update`.
-struct PersistenceNotifier {
- /// Users won't access the persistence_lock directly, but rather wait on its bool using
- /// `wait_timeout` and `wait`.
- persistence_lock: (Mutex<bool>, Condvar),
-}
-
-impl PersistenceNotifier {
- fn new() -> Self {
- Self {
- persistence_lock: (Mutex::new(false), Condvar::new()),
- }
- }
-
- fn wait(&self) {
- loop {
- let &(ref mtx, ref cvar) = &self.persistence_lock;
- let mut guard = mtx.lock().unwrap();
- if *guard {
- *guard = false;
- return;
- }
- guard = cvar.wait(guard).unwrap();
- let result = *guard;
- if result {
- *guard = false;
- return
- }
- }
- }
-
- #[cfg(any(test, feature = "std"))]
- fn wait_timeout(&self, max_wait: Duration) -> bool {
- let current_time = Instant::now();
- loop {
- let &(ref mtx, ref cvar) = &self.persistence_lock;
- let mut guard = mtx.lock().unwrap();
- if *guard {
- *guard = false;
- return true;
- }
- guard = cvar.wait_timeout(guard, max_wait).unwrap().0;
- // Due to spurious wakeups that can happen on `wait_timeout`, here we need to check if the
- // desired wait time has actually passed, and if not then restart the loop with a reduced wait
- // time. Note that this logic can be highly simplified through the use of
- // `Condvar::wait_while` and `Condvar::wait_timeout_while`, if and when our MSRV is raised to
- // 1.42.0.
- let elapsed = current_time.elapsed();
- let result = *guard;
- if result || elapsed >= max_wait {
- *guard = false;
- return result;
- }
- match max_wait.checked_sub(elapsed) {
- None => return result,
- Some(_) => continue
- }
- }
- }
-
- // Signal to the ChannelManager persister that there are updates necessitating persisting to disk.
- fn notify(&self) {
- let &(ref persist_mtx, ref cnd) = &self.persistence_lock;
- let mut persistence_lock = persist_mtx.lock().unwrap();
- *persistence_lock = true;
- mem::drop(persistence_lock);
- cnd.notify_all();
- }
-}
-
const SERIALIZATION_VERSION: u8 = 1;
const MIN_SERIALIZATION_VERSION: u8 = 1;
pending_events: Mutex::new(pending_events_read),
pending_background_events: Mutex::new(pending_background_events_read),
total_consistency_lock: RwLock::new(()),
- persistence_notifier: PersistenceNotifier::new(),
+ persistence_notifier: Notifier::new(),
keys_manager: args.keys_manager,
logger: args.logger,
use util::test_utils;
use chain::keysinterface::KeysInterface;
- #[cfg(feature = "std")]
- #[test]
- fn test_wait_timeout() {
- use ln::channelmanager::PersistenceNotifier;
- use sync::Arc;
- use core::sync::atomic::AtomicBool;
- use std::thread;
-
- let persistence_notifier = Arc::new(PersistenceNotifier::new());
- let thread_notifier = Arc::clone(&persistence_notifier);
-
- let exit_thread = Arc::new(AtomicBool::new(false));
- let exit_thread_clone = exit_thread.clone();
- thread::spawn(move || {
- loop {
- let &(ref persist_mtx, ref cnd) = &thread_notifier.persistence_lock;
- let mut persistence_lock = persist_mtx.lock().unwrap();
- *persistence_lock = true;
- cnd.notify_all();
-
- if exit_thread_clone.load(Ordering::SeqCst) {
- break
- }
- }
- });
-
- // Check that we can block indefinitely until updates are available.
- let _ = persistence_notifier.wait();
-
- // Check that the PersistenceNotifier will return after the given duration if updates are
- // available.
- loop {
- if persistence_notifier.wait_timeout(Duration::from_millis(100)) {
- break
- }
- }
-
- exit_thread.store(true, Ordering::SeqCst);
-
- // Check that the PersistenceNotifier will return after the given duration even if no updates
- // are available.
- loop {
- if !persistence_notifier.wait_timeout(Duration::from_millis(100)) {
- break
- }
- }
- }
-
#[test]
fn test_notify_limits() {
// Check that a few cases which don't require the persistence of a new ChannelManager,
projects / rust-lightning / blobdiff