}}
}
+macro_rules! emit_channel_pending_event {
+ ($locked_events: expr, $channel: expr) => {
+ if $channel.should_emit_channel_pending_event() {
+ $locked_events.push(events::Event::ChannelPending {
+ channel_id: $channel.channel_id(),
+ former_temporary_channel_id: $channel.temporary_channel_id(),
+ counterparty_node_id: $channel.get_counterparty_node_id(),
+ user_channel_id: $channel.get_user_id(),
+ funding_txo: $channel.get_funding_txo().unwrap().into_bitcoin_outpoint(),
+ });
+ $channel.set_channel_pending_event_emitted();
+ }
+ }
+}
+
macro_rules! emit_channel_ready_event {
- ($self: expr, $channel: expr) => {
+ ($locked_events: expr, $channel: expr) => {
if $channel.should_emit_channel_ready_event() {
- {
- let mut pending_events = $self.pending_events.lock().unwrap();
- pending_events.push(events::Event::ChannelReady {
- channel_id: $channel.channel_id(),
- user_channel_id: $channel.get_user_id(),
- counterparty_node_id: $channel.get_counterparty_node_id(),
- channel_type: $channel.get_channel_type().clone(),
- });
- }
+ debug_assert!($channel.channel_pending_event_emitted());
+ $locked_events.push(events::Event::ChannelReady {
+ channel_id: $channel.channel_id(),
+ user_channel_id: $channel.get_user_id(),
+ counterparty_node_id: $channel.get_counterparty_node_id(),
+ channel_type: $channel.get_channel_type().clone(),
+ });
$channel.set_channel_ready_event_emitted();
}
}
}
}
let mut total_value = claimable_htlc.sender_intended_value;
+ let mut earliest_expiry = claimable_htlc.cltv_expiry;
for htlc in htlcs.iter() {
total_value += htlc.sender_intended_value;
+ earliest_expiry = cmp::min(earliest_expiry, htlc.cltv_expiry);
match &htlc.onion_payload {
OnionPayload::Invoice { .. } => {
if htlc.total_msat != $payment_data.total_msat {
amount_msat,
via_channel_id: Some(prev_channel_id),
via_user_channel_id: Some(prev_user_channel_id),
+ claim_deadline: Some(earliest_expiry - HTLC_FAIL_BACK_BUFFER),
});
payment_claimable_generated = true;
} else {
hash_map::Entry::Vacant(e) => {
let amount_msat = claimable_htlc.value;
claimable_htlc.total_value_received = Some(amount_msat);
+ let claim_deadline = Some(claimable_htlc.cltv_expiry - HTLC_FAIL_BACK_BUFFER);
let purpose = events::PaymentPurpose::SpontaneousPayment(preimage);
e.insert((purpose.clone(), vec![claimable_htlc]));
let prev_channel_id = prev_funding_outpoint.to_channel_id();
purpose,
via_channel_id: Some(prev_channel_id),
via_user_channel_id: Some(prev_user_channel_id),
+ claim_deadline,
});
},
hash_map::Entry::Occupied(_) => {
/// Provides a payment preimage in response to [`Event::PaymentClaimable`], generating any
/// [`MessageSendEvent`]s needed to claim the payment.
///
- /// Note that calling this method does *not* guarantee that the payment has been claimed. You
- /// *must* wait for an [`Event::PaymentClaimed`] event which upon a successful claim will be
- /// provided to your [`EventHandler`] when [`process_pending_events`] is next called.
+ /// This method is guaranteed to ensure the payment has been claimed but only if the current
+ /// height is strictly below [`Event::PaymentClaimable::claim_deadline`]. To avoid race
+ /// conditions, you should wait for an [`Event::PaymentClaimed`] before considering the payment
+ /// successful. It will generally be available in the next [`process_pending_events`] call.
///
/// Note that if you did not set an `amount_msat` when calling [`create_inbound_payment`] or
/// [`create_inbound_payment_for_hash`] you must check that the amount in the `PaymentClaimable`
/// the sender "proof-of-payment" when they did not fulfill the full expected payment.
///
/// [`Event::PaymentClaimable`]: crate::events::Event::PaymentClaimable
+ /// [`Event::PaymentClaimable::claim_deadline`]: crate::events::Event::PaymentClaimable::claim_deadline
/// [`Event::PaymentClaimed`]: crate::events::Event::PaymentClaimed
/// [`process_pending_events`]: EventsProvider::process_pending_events
/// [`create_inbound_payment`]: Self::create_inbound_payment
};
debug_assert!(!sources.is_empty());
- // If we are claiming an MPP payment, we check that all channels which contain a claimable
- // HTLC still exist. While this isn't guaranteed to remain true if a channel closes while
- // we're claiming (or even after we claim, before the commitment update dance completes),
- // it should be a relatively rare race, and we'd rather not claim HTLCs that require us to
- // go on-chain (and lose the on-chain fee to do so) than just reject the payment.
- //
- // Note that we'll still always get our funds - as long as the generated
- // `ChannelMonitorUpdate` makes it out to the relevant monitor we can claim on-chain.
- //
- // If we find an HTLC which we would need to claim but for which we do not have a
- // channel, we will fail all parts of the MPP payment. While we could wait and see if
- // the sender retries the already-failed path(s), it should be a pretty rare case where
- // we got all the HTLCs and then a channel closed while we were waiting for the user to
- // provide the preimage, so worrying too much about the optimal handling isn't worth
- // it.
+ // Just in case one HTLC has been failed between when we generated the `PaymentClaimable`
+ // and when we got here we need to check that the amount we're about to claim matches the
+ // amount we told the user in the last `PaymentClaimable`. We also do a sanity-check that
+ // the MPP parts all have the same `total_msat`.
let mut claimable_amt_msat = 0;
let mut prev_total_msat = None;
let mut expected_amt_msat = None;
let mut errs = Vec::new();
let per_peer_state = self.per_peer_state.read().unwrap();
for htlc in sources.iter() {
- let (counterparty_node_id, chan_id) = match self.short_to_chan_info.read().unwrap().get(&htlc.prev_hop.short_channel_id) {
- Some((cp_id, chan_id)) => (cp_id.clone(), chan_id.clone()),
- None => {
- valid_mpp = false;
- break;
- }
- };
-
- let peer_state_mutex_opt = per_peer_state.get(&counterparty_node_id);
- if peer_state_mutex_opt.is_none() {
- valid_mpp = false;
- break;
- }
-
- let mut peer_state_lock = peer_state_mutex_opt.unwrap().lock().unwrap();
- let peer_state = &mut *peer_state_lock;
-
- if peer_state.channel_by_id.get(&chan_id).is_none() {
- valid_mpp = false;
- break;
- }
-
if prev_total_msat.is_some() && prev_total_msat != Some(htlc.total_msat) {
log_error!(self.logger, "Somehow ended up with an MPP payment with different expected total amounts - this should not be reachable!");
debug_assert!(false);
});
}
- emit_channel_ready_event!(self, channel);
-
macro_rules! handle_cs { () => {
if let Some(update) = commitment_update {
pending_msg_events.push(events::MessageSendEvent::UpdateHTLCs {
self.tx_broadcaster.broadcast_transaction(&tx);
}
+ {
+ let mut pending_events = self.pending_events.lock().unwrap();
+ emit_channel_pending_event!(pending_events, channel);
+ emit_channel_ready_event!(pending_events, channel);
+ }
+
htlc_forwards
}
}
}
- emit_channel_ready_event!(self, chan.get_mut());
+ {
+ let mut pending_events = self.pending_events.lock().unwrap();
+ emit_channel_ready_event!(pending_events, chan.get_mut());
+ }
Ok(())
},
}
}
- emit_channel_ready_event!(self, channel);
+ {
+ let mut pending_events = self.pending_events.lock().unwrap();
+ emit_channel_ready_event!(pending_events, channel);
+ }
if let Some(announcement_sigs) = announcement_sigs {
log_trace!(self.logger, "Sending announcement_signatures for channel {}", log_bytes!(channel.channel_id()));
}
}
- /// Blocks until ChannelManager needs to be persisted or a timeout is reached. It returns a bool
- /// indicating whether persistence is necessary. Only one listener on
- /// [`await_persistable_update`], [`await_persistable_update_timeout`], or a future returned by
- /// [`get_persistable_update_future`] is guaranteed to be woken up.
- ///
- /// Note that this method is not available with the `no-std` feature.
+ /// Gets a [`Future`] that completes when this [`ChannelManager`] needs to be persisted.
///
- /// [`await_persistable_update`]: Self::await_persistable_update
- /// [`await_persistable_update_timeout`]: Self::await_persistable_update_timeout
- /// [`get_persistable_update_future`]: Self::get_persistable_update_future
- #[cfg(any(test, feature = "std"))]
- pub fn await_persistable_update_timeout(&self, max_wait: Duration) -> bool {
- self.persistence_notifier.wait_timeout(max_wait)
- }
-
- /// Blocks until ChannelManager needs to be persisted. Only one listener on
- /// [`await_persistable_update`], `await_persistable_update_timeout`, or a future returned by
- /// [`get_persistable_update_future`] is guaranteed to be woken up.
+ /// Note that callbacks registered on the [`Future`] MUST NOT call back into this
+ /// [`ChannelManager`] and should instead register actions to be taken later.
///
- /// [`await_persistable_update`]: Self::await_persistable_update
- /// [`get_persistable_update_future`]: Self::get_persistable_update_future
- pub fn await_persistable_update(&self) {
- 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()
}
let mut id_to_peer = HashMap::with_capacity(cmp::min(channel_count as usize, 128));
let mut short_to_chan_info = HashMap::with_capacity(cmp::min(channel_count as usize, 128));
let mut channel_closures = Vec::new();
+ let mut pending_background_events = Vec::new();
for _ in 0..channel_count {
let mut channel: Channel<<SP::Target as SignerProvider>::Signer> = Channel::read(reader, (
&args.entropy_source, &args.signer_provider, best_block_height, &provided_channel_type_features(&args.default_config)
log_error!(args.logger, " The channel will be force-closed and the latest commitment transaction from the ChannelMonitor broadcast.");
log_error!(args.logger, " The ChannelMonitor for channel {} is at update_id {} but the ChannelManager is at update_id {}.",
log_bytes!(channel.channel_id()), monitor.get_latest_update_id(), channel.get_latest_monitor_update_id());
- let (_, mut new_failed_htlcs) = channel.force_shutdown(true);
+ let (monitor_update, mut new_failed_htlcs) = channel.force_shutdown(true);
+ if let Some(monitor_update) = monitor_update {
+ pending_background_events.push(BackgroundEvent::ClosingMonitorUpdate(monitor_update));
+ }
failed_htlcs.append(&mut new_failed_htlcs);
- monitor.broadcast_latest_holder_commitment_txn(&args.tx_broadcaster, &args.logger);
channel_closures.push(events::Event::ChannelClosed {
channel_id: channel.channel_id(),
user_channel_id: channel.get_user_id(),
}
}
- for (funding_txo, monitor) in args.channel_monitors.iter_mut() {
+ for (funding_txo, _) in args.channel_monitors.iter() {
if !funding_txo_set.contains(funding_txo) {
- log_info!(args.logger, "Broadcasting latest holder commitment transaction for closed channel {}", log_bytes!(funding_txo.to_channel_id()));
- monitor.broadcast_latest_holder_commitment_txn(&args.tx_broadcaster, &args.logger);
+ let monitor_update = ChannelMonitorUpdate {
+ update_id: CLOSED_CHANNEL_UPDATE_ID,
+ updates: vec![ChannelMonitorUpdateStep::ChannelForceClosed { should_broadcast: true }],
+ };
+ pending_background_events.push(BackgroundEvent::ClosingMonitorUpdate((*funding_txo, monitor_update)));
}
}
}
let background_event_count: u64 = Readable::read(reader)?;
- let mut pending_background_events_read: Vec<BackgroundEvent> = Vec::with_capacity(cmp::min(background_event_count as usize, MAX_ALLOC_SIZE/mem::size_of::<BackgroundEvent>()));
for _ in 0..background_event_count {
match <u8 as Readable>::read(reader)? {
- 0 => pending_background_events_read.push(BackgroundEvent::ClosingMonitorUpdate((Readable::read(reader)?, Readable::read(reader)?))),
+ 0 => {
+ let (funding_txo, monitor_update): (OutPoint, ChannelMonitorUpdate) = (Readable::read(reader)?, Readable::read(reader)?);
+ if pending_background_events.iter().find(|e| {
+ let BackgroundEvent::ClosingMonitorUpdate((pending_funding_txo, pending_monitor_update)) = e;
+ *pending_funding_txo == funding_txo && *pending_monitor_update == monitor_update
+ }).is_none() {
+ pending_background_events.push(BackgroundEvent::ClosingMonitorUpdate((funding_txo, monitor_update)));
+ }
+ }
_ => return Err(DecodeError::InvalidValue),
}
}
per_peer_state: FairRwLock::new(per_peer_state),
pending_events: Mutex::new(pending_events_read),
- pending_background_events: Mutex::new(pending_background_events_read),
+ pending_background_events: Mutex::new(pending_background_events),
total_consistency_lock: RwLock::new(()),
persistence_notifier: Notifier::new(),
use bitcoin::hashes::Hash;
use bitcoin::hashes::sha256::Hash as Sha256;
use bitcoin::secp256k1::{PublicKey, Secp256k1, SecretKey};
+ #[cfg(feature = "std")]
use core::time::Duration;
use core::sync::atomic::Ordering;
use crate::events::{Event, HTLCDestination, MessageSendEvent, MessageSendEventsProvider, ClosureReason};
// All nodes start with a persistable update pending as `create_network` connects each node
// with all other nodes to make most tests simpler.
- assert!(nodes[0].node.await_persistable_update_timeout(Duration::from_millis(1)));
- assert!(nodes[1].node.await_persistable_update_timeout(Duration::from_millis(1)));
- assert!(nodes[2].node.await_persistable_update_timeout(Duration::from_millis(1)));
+ assert!(nodes[0].node.get_persistable_update_future().poll_is_complete());
+ assert!(nodes[1].node.get_persistable_update_future().poll_is_complete());
+ assert!(nodes[2].node.get_persistable_update_future().poll_is_complete());
let mut chan = create_announced_chan_between_nodes(&nodes, 0, 1);
&nodes[0].node.get_our_node_id()).pop().unwrap();
// The first two nodes (which opened a channel) should now require fresh persistence
- assert!(nodes[0].node.await_persistable_update_timeout(Duration::from_millis(1)));
- assert!(nodes[1].node.await_persistable_update_timeout(Duration::from_millis(1)));
+ assert!(nodes[0].node.get_persistable_update_future().poll_is_complete());
+ assert!(nodes[1].node.get_persistable_update_future().poll_is_complete());
// ... but the last node should not.
- assert!(!nodes[2].node.await_persistable_update_timeout(Duration::from_millis(1)));
+ assert!(!nodes[2].node.get_persistable_update_future().poll_is_complete());
// After persisting the first two nodes they should no longer need fresh persistence.
- assert!(!nodes[0].node.await_persistable_update_timeout(Duration::from_millis(1)));
- assert!(!nodes[1].node.await_persistable_update_timeout(Duration::from_millis(1)));
+ assert!(!nodes[0].node.get_persistable_update_future().poll_is_complete());
+ assert!(!nodes[1].node.get_persistable_update_future().poll_is_complete());
// Node 3, unrelated to the only channel, shouldn't care if it receives a channel_update
// about the channel.
nodes[2].node.handle_channel_update(&nodes[1].node.get_our_node_id(), &chan.0);
nodes[2].node.handle_channel_update(&nodes[1].node.get_our_node_id(), &chan.1);
- assert!(!nodes[2].node.await_persistable_update_timeout(Duration::from_millis(1)));
+ assert!(!nodes[2].node.get_persistable_update_future().poll_is_complete());
// The nodes which are a party to the channel should also ignore messages from unrelated
// parties.
nodes[0].node.handle_channel_update(&nodes[2].node.get_our_node_id(), &chan.1);
nodes[1].node.handle_channel_update(&nodes[2].node.get_our_node_id(), &chan.0);
nodes[1].node.handle_channel_update(&nodes[2].node.get_our_node_id(), &chan.1);
- assert!(!nodes[0].node.await_persistable_update_timeout(Duration::from_millis(1)));
- assert!(!nodes[1].node.await_persistable_update_timeout(Duration::from_millis(1)));
+ assert!(!nodes[0].node.get_persistable_update_future().poll_is_complete());
+ assert!(!nodes[1].node.get_persistable_update_future().poll_is_complete());
// At this point the channel info given by peers should still be the same.
assert_eq!(nodes[0].node.list_channels()[0], node_a_chan_info);
// persisted and that its channel info remains the same.
nodes[0].node.handle_channel_update(&nodes[1].node.get_our_node_id(), &as_update);
nodes[1].node.handle_channel_update(&nodes[0].node.get_our_node_id(), &bs_update);
- assert!(!nodes[0].node.await_persistable_update_timeout(Duration::from_millis(1)));
- assert!(!nodes[1].node.await_persistable_update_timeout(Duration::from_millis(1)));
+ assert!(!nodes[0].node.get_persistable_update_future().poll_is_complete());
+ assert!(!nodes[1].node.get_persistable_update_future().poll_is_complete());
assert_eq!(nodes[0].node.list_channels()[0], node_a_chan_info);
assert_eq!(nodes[1].node.list_channels()[0], node_b_chan_info);
// the channel info has updated.
nodes[0].node.handle_channel_update(&nodes[1].node.get_our_node_id(), &bs_update);
nodes[1].node.handle_channel_update(&nodes[0].node.get_our_node_id(), &as_update);
- assert!(nodes[0].node.await_persistable_update_timeout(Duration::from_millis(1)));
- assert!(nodes[1].node.await_persistable_update_timeout(Duration::from_millis(1)));
+ assert!(nodes[0].node.get_persistable_update_future().poll_is_complete());
+ assert!(nodes[1].node.get_persistable_update_future().poll_is_complete());
assert_ne!(nodes[0].node.list_channels()[0], node_a_chan_info);
assert_ne!(nodes[1].node.list_channels()[0], node_b_chan_info);
}
assert_eq!(nodes_0_lock.len(), 1);
assert!(nodes_0_lock.contains_key(channel_id));
}
+ expect_channel_pending_event(&nodes[1], &nodes[0].node.get_our_node_id());
{
// Assert that `nodes[1]`'s `id_to_peer` map is populated with the channel as soon as
let funding_signed = get_event_msg!(nodes[1], MessageSendEvent::SendFundingSigned, nodes[0].node.get_our_node_id());
nodes[0].node.handle_funding_signed(&nodes[1].node.get_our_node_id(), &funding_signed);
check_added_monitors!(nodes[0], 1);
+ expect_channel_pending_event(&nodes[0], &nodes[1].node.get_our_node_id());
let (channel_ready, _) = create_chan_between_nodes_with_value_confirm(&nodes[0], &nodes[1], &tx);
let (announcement, nodes_0_update, nodes_1_update) = create_chan_between_nodes_with_value_b(&nodes[0], &nodes[1], &channel_ready);
update_nodes_with_chan_announce(&nodes, 0, 1, &announcement, &nodes_0_update, &nodes_1_update);
nodes[1].node.handle_funding_created(&nodes[0].node.get_our_node_id(), &funding_created_msg);
check_added_monitors!(nodes[1], 1);
+ expect_channel_pending_event(&nodes[1], &nodes[0].node.get_our_node_id());
+
let funding_signed = get_event_msg!(nodes[1], MessageSendEvent::SendFundingSigned, nodes[0].node.get_our_node_id());
nodes[0].node.handle_funding_signed(&nodes[1].node.get_our_node_id(), &funding_signed);
check_added_monitors!(nodes[0], 1);
+ expect_channel_pending_event(&nodes[0], &nodes[1].node.get_our_node_id());
}
open_channel_msg.temporary_channel_id = nodes[0].keys_manager.get_secure_random_bytes();
}
use crate::chain::chainmonitor::{ChainMonitor, Persist};
use crate::chain::keysinterface::{EntropySource, KeysManager, InMemorySigner};
use crate::events::{Event, MessageSendEvent, MessageSendEventsProvider};
- use crate::ln::channelmanager::{self, BestBlock, ChainParameters, ChannelManager, PaymentHash, PaymentPreimage, PaymentId};
+ use crate::ln::channelmanager::{BestBlock, ChainParameters, ChannelManager, PaymentHash, PaymentPreimage, PaymentId};
use crate::ln::functional_test_utils::*;
use crate::ln::msgs::{ChannelMessageHandler, Init};
use crate::routing::gossip::NetworkGraph;
} else { panic!(); }
node_b.handle_funding_created(&node_a.get_our_node_id(), &get_event_msg!(node_a_holder, MessageSendEvent::SendFundingCreated, node_b.get_our_node_id()));
+ let events_b = node_b.get_and_clear_pending_events();
+ assert_eq!(events_b.len(), 1);
+ match events_b[0] {
+ Event::ChannelPending{ ref counterparty_node_id, .. } => {
+ assert_eq!(*counterparty_node_id, node_a.get_our_node_id());
+ },
+ _ => panic!("Unexpected event"),
+ }
+
node_a.handle_funding_signed(&node_b.get_our_node_id(), &get_event_msg!(node_b_holder, MessageSendEvent::SendFundingSigned, node_a.get_our_node_id()));
+ let events_a = node_a.get_and_clear_pending_events();
+ assert_eq!(events_a.len(), 1);
+ match events_a[0] {
+ Event::ChannelPending{ ref counterparty_node_id, .. } => {
+ assert_eq!(*counterparty_node_id, node_b.get_our_node_id());
+ },
+ _ => panic!("Unexpected event"),
+ }
assert_eq!(&tx_broadcaster.txn_broadcasted.lock().unwrap()[..], &[tx.clone()]);
projects / rust-lightning / blobdiff