diff --git a/lightning/src/chain/channelmonitor.rs b/lightning/src/chain/channelmonitor.rs index 7fad40d9c2e..8be785f29d2 100644 --- a/lightning/src/chain/channelmonitor.rs +++ b/lightning/src/chain/channelmonitor.rs @@ -1515,6 +1515,101 @@ impl ChannelMonitor { res } + + /// Gets the set of outbound HTLCs which are pending resolution in this channel. + /// This is used to reconstruct pending outbound payments on restart in the ChannelManager. + pub(crate) fn get_pending_outbound_htlcs(&self) -> HashMap { + let mut res = HashMap::new(); + let us = self.inner.lock().unwrap(); + + macro_rules! walk_htlcs { + ($holder_commitment: expr, $htlc_iter: expr) => { + for (htlc, source) in $htlc_iter { + if us.htlcs_resolved_on_chain.iter().any(|v| Some(v.input_idx) == htlc.transaction_output_index) { + // We should assert that funding_spend_confirmed is_some() here, but we + // have some unit tests which violate HTLC transaction CSVs entirely and + // would fail. + // TODO: Once tests all connect transactions at consensus-valid times, we + // should assert here like we do in `get_claimable_balances`. + } else if htlc.offered == $holder_commitment { + // If the payment was outbound, check if there's an HTLCUpdate + // indicating we have spent this HTLC with a timeout, claiming it back + // and awaiting confirmations on it. + let htlc_update_confd = us.onchain_events_awaiting_threshold_conf.iter().any(|event| { + if let OnchainEvent::HTLCUpdate { input_idx: Some(input_idx), .. } = event.event { + // If the HTLC was timed out, we wait for ANTI_REORG_DELAY blocks + // before considering it "no longer pending" - this matches when we + // provide the ChannelManager an HTLC failure event. + Some(input_idx) == htlc.transaction_output_index && + us.best_block.height() >= event.height + ANTI_REORG_DELAY - 1 + } else if let OnchainEvent::HTLCSpendConfirmation { input_idx, .. } = event.event { + // If the HTLC was fulfilled with a preimage, we consider the HTLC + // immediately non-pending, matching when we provide ChannelManager + // the preimage. + Some(input_idx) == htlc.transaction_output_index + } else { false } + }); + if !htlc_update_confd { + res.insert(source.clone(), htlc.clone()); + } + } + } + } + } + + // We're only concerned with the confirmation count of HTLC transactions, and don't + // actually care how many confirmations a commitment transaction may or may not have. Thus, + // we look for either a FundingSpendConfirmation event or a funding_spend_confirmed. + let confirmed_txid = us.funding_spend_confirmed.or_else(|| { + us.onchain_events_awaiting_threshold_conf.iter().find_map(|event| { + if let OnchainEvent::FundingSpendConfirmation { .. } = event.event { + Some(event.txid) + } else { None } + }) + }); + if let Some(txid) = confirmed_txid { + if Some(txid) == us.current_counterparty_commitment_txid || Some(txid) == us.prev_counterparty_commitment_txid { + walk_htlcs!(false, us.counterparty_claimable_outpoints.get(&txid).unwrap().iter().filter_map(|(a, b)| { + if let &Some(ref source) = b { + Some((a, &**source)) + } else { None } + })); + } else if txid == us.current_holder_commitment_tx.txid { + walk_htlcs!(true, us.current_holder_commitment_tx.htlc_outputs.iter().filter_map(|(a, _, c)| { + if let Some(source) = c { Some((a, source)) } else { None } + })); + } else if let Some(prev_commitment) = &us.prev_holder_signed_commitment_tx { + if txid == prev_commitment.txid { + walk_htlcs!(true, prev_commitment.htlc_outputs.iter().filter_map(|(a, _, c)| { + if let Some(source) = c { Some((a, source)) } else { None } + })); + } + } + } else { + // If we have not seen a commitment transaction on-chain (ie the channel is not yet + // closed), just examine the available counterparty commitment transactions. See docs + // on `fail_unbroadcast_htlcs`, below, for justification. + macro_rules! walk_counterparty_commitment { + ($txid: expr) => { + if let Some(ref latest_outpoints) = us.counterparty_claimable_outpoints.get($txid) { + for &(ref htlc, ref source_option) in latest_outpoints.iter() { + if let &Some(ref source) = source_option { + res.insert((**source).clone(), htlc.clone()); + } + } + } + } + } + if let Some(ref txid) = us.current_counterparty_commitment_txid { + walk_counterparty_commitment!(txid); + } + if let Some(ref txid) = us.prev_counterparty_commitment_txid { + walk_counterparty_commitment!(txid); + } + } + + res + } } /// Compares a broadcasted commitment transaction's HTLCs with those in the latest state, diff --git a/lightning/src/ln/channel.rs b/lightning/src/ln/channel.rs index 7883d7b29e4..c2e158627ad 100644 --- a/lightning/src/ln/channel.rs +++ b/lightning/src/ln/channel.rs @@ -339,6 +339,29 @@ pub enum UpdateFulfillCommitFetch { DuplicateClaim {}, } +/// The return value of `revoke_and_ack` on success, primarily updates to other channels or HTLC +/// state. +pub(super) struct RAAUpdates { + pub commitment_update: Option, + pub accepted_htlcs: Vec<(PendingHTLCInfo, u64)>, + pub failed_htlcs: Vec<(HTLCSource, PaymentHash, HTLCFailReason)>, + pub finalized_claimed_htlcs: Vec, + pub monitor_update: ChannelMonitorUpdate, + pub holding_cell_failed_htlcs: Vec<(HTLCSource, PaymentHash)>, +} + +/// The return value of `monitor_updating_restored` +pub(super) struct MonitorRestoreUpdates { + pub raa: Option, + pub commitment_update: Option, + pub order: RAACommitmentOrder, + pub accepted_htlcs: Vec<(PendingHTLCInfo, u64)>, + pub failed_htlcs: Vec<(HTLCSource, PaymentHash, HTLCFailReason)>, + pub finalized_claimed_htlcs: Vec, + pub funding_broadcastable: Option, + pub funding_locked: Option, +} + /// If the majority of the channels funds are to the fundee and the initiator holds only just /// enough funds to cover their reserve value, channels are at risk of getting "stuck". Because the /// initiator controls the feerate, if they then go to increase the channel fee, they may have no @@ -406,6 +429,7 @@ pub(super) struct Channel { monitor_pending_commitment_signed: bool, monitor_pending_forwards: Vec<(PendingHTLCInfo, u64)>, monitor_pending_failures: Vec<(HTLCSource, PaymentHash, HTLCFailReason)>, + monitor_pending_finalized_fulfills: Vec, // pending_update_fee is filled when sending and receiving update_fee. // @@ -692,6 +716,7 @@ impl Channel { monitor_pending_commitment_signed: false, monitor_pending_forwards: Vec::new(), monitor_pending_failures: Vec::new(), + monitor_pending_finalized_fulfills: Vec::new(), #[cfg(debug_assertions)] holder_max_commitment_tx_output: Mutex::new((channel_value_satoshis * 1000 - push_msat, push_msat)), @@ -955,6 +980,7 @@ impl Channel { monitor_pending_commitment_signed: false, monitor_pending_forwards: Vec::new(), monitor_pending_failures: Vec::new(), + monitor_pending_finalized_fulfills: Vec::new(), #[cfg(debug_assertions)] holder_max_commitment_tx_output: Mutex::new((msg.push_msat, msg.funding_satoshis * 1000 - msg.push_msat)), @@ -2711,7 +2737,7 @@ impl Channel { /// waiting on this revoke_and_ack. The generation of this new commitment_signed may also fail, /// generating an appropriate error *after* the channel state has been updated based on the /// revoke_and_ack message. - pub fn revoke_and_ack(&mut self, msg: &msgs::RevokeAndACK, logger: &L) -> Result<(Option, Vec<(PendingHTLCInfo, u64)>, Vec<(HTLCSource, PaymentHash, HTLCFailReason)>, ChannelMonitorUpdate, Vec<(HTLCSource, PaymentHash)>), ChannelError> + pub fn revoke_and_ack(&mut self, msg: &msgs::RevokeAndACK, logger: &L) -> Result where L::Target: Logger, { if (self.channel_state & (ChannelState::ChannelFunded as u32)) != (ChannelState::ChannelFunded as u32) { @@ -2777,6 +2803,7 @@ impl Channel { log_trace!(logger, "Updating HTLCs on receipt of RAA in channel {}...", log_bytes!(self.channel_id())); let mut to_forward_infos = Vec::new(); let mut revoked_htlcs = Vec::new(); + let mut finalized_claimed_htlcs = Vec::new(); let mut update_fail_htlcs = Vec::new(); let mut update_fail_malformed_htlcs = Vec::new(); let mut require_commitment = false; @@ -2803,6 +2830,7 @@ impl Channel { if let Some(reason) = fail_reason.clone() { // We really want take() here, but, again, non-mut ref :( revoked_htlcs.push((htlc.source.clone(), htlc.payment_hash, reason)); } else { + finalized_claimed_htlcs.push(htlc.source.clone()); // They fulfilled, so we sent them money value_to_self_msat_diff -= htlc.amount_msat as i64; } @@ -2899,8 +2927,14 @@ impl Channel { } self.monitor_pending_forwards.append(&mut to_forward_infos); self.monitor_pending_failures.append(&mut revoked_htlcs); + self.monitor_pending_finalized_fulfills.append(&mut finalized_claimed_htlcs); log_debug!(logger, "Received a valid revoke_and_ack for channel {} but awaiting a monitor update resolution to reply.", log_bytes!(self.channel_id())); - return Ok((None, Vec::new(), Vec::new(), monitor_update, Vec::new())) + return Ok(RAAUpdates { + commitment_update: None, finalized_claimed_htlcs: Vec::new(), + accepted_htlcs: Vec::new(), failed_htlcs: Vec::new(), + monitor_update, + holding_cell_failed_htlcs: Vec::new() + }); } match self.free_holding_cell_htlcs(logger)? { @@ -2919,7 +2953,14 @@ impl Channel { self.latest_monitor_update_id = monitor_update.update_id; monitor_update.updates.append(&mut additional_update.updates); - Ok((Some(commitment_update), to_forward_infos, revoked_htlcs, monitor_update, htlcs_to_fail)) + Ok(RAAUpdates { + commitment_update: Some(commitment_update), + finalized_claimed_htlcs, + accepted_htlcs: to_forward_infos, + failed_htlcs: revoked_htlcs, + monitor_update, + holding_cell_failed_htlcs: htlcs_to_fail + }) }, (None, htlcs_to_fail) => { if require_commitment { @@ -2932,17 +2973,27 @@ impl Channel { log_debug!(logger, "Received a valid revoke_and_ack for channel {}. Responding with a commitment update with {} HTLCs failed.", log_bytes!(self.channel_id()), update_fail_htlcs.len() + update_fail_malformed_htlcs.len()); - Ok((Some(msgs::CommitmentUpdate { - update_add_htlcs: Vec::new(), - update_fulfill_htlcs: Vec::new(), - update_fail_htlcs, - update_fail_malformed_htlcs, - update_fee: None, - commitment_signed - }), to_forward_infos, revoked_htlcs, monitor_update, htlcs_to_fail)) + Ok(RAAUpdates { + commitment_update: Some(msgs::CommitmentUpdate { + update_add_htlcs: Vec::new(), + update_fulfill_htlcs: Vec::new(), + update_fail_htlcs, + update_fail_malformed_htlcs, + update_fee: None, + commitment_signed + }), + finalized_claimed_htlcs, + accepted_htlcs: to_forward_infos, failed_htlcs: revoked_htlcs, + monitor_update, holding_cell_failed_htlcs: htlcs_to_fail + }) } else { log_debug!(logger, "Received a valid revoke_and_ack for channel {} with no reply necessary.", log_bytes!(self.channel_id())); - Ok((None, to_forward_infos, revoked_htlcs, monitor_update, htlcs_to_fail)) + Ok(RAAUpdates { + commitment_update: None, + finalized_claimed_htlcs, + accepted_htlcs: to_forward_infos, failed_htlcs: revoked_htlcs, + monitor_update, holding_cell_failed_htlcs: htlcs_to_fail + }) } } } @@ -3057,18 +3108,23 @@ impl Channel { /// which failed. The messages which were generated from that call which generated the /// monitor update failure must *not* have been sent to the remote end, and must instead /// have been dropped. They will be regenerated when monitor_updating_restored is called. - pub fn monitor_update_failed(&mut self, resend_raa: bool, resend_commitment: bool, mut pending_forwards: Vec<(PendingHTLCInfo, u64)>, mut pending_fails: Vec<(HTLCSource, PaymentHash, HTLCFailReason)>) { + pub fn monitor_update_failed(&mut self, resend_raa: bool, resend_commitment: bool, + mut pending_forwards: Vec<(PendingHTLCInfo, u64)>, + mut pending_fails: Vec<(HTLCSource, PaymentHash, HTLCFailReason)>, + mut pending_finalized_claimed_htlcs: Vec + ) { self.monitor_pending_revoke_and_ack |= resend_raa; self.monitor_pending_commitment_signed |= resend_commitment; self.monitor_pending_forwards.append(&mut pending_forwards); self.monitor_pending_failures.append(&mut pending_fails); + self.monitor_pending_finalized_fulfills.append(&mut pending_finalized_claimed_htlcs); self.channel_state |= ChannelState::MonitorUpdateFailed as u32; } /// Indicates that the latest ChannelMonitor update has been committed by the client /// successfully and we should restore normal operation. Returns messages which should be sent /// to the remote side. - pub fn monitor_updating_restored(&mut self, logger: &L) -> (Option, Option, RAACommitmentOrder, Vec<(PendingHTLCInfo, u64)>, Vec<(HTLCSource, PaymentHash, HTLCFailReason)>, Option, Option) where L::Target: Logger { + pub fn monitor_updating_restored(&mut self, logger: &L) -> MonitorRestoreUpdates where L::Target: Logger { assert_eq!(self.channel_state & ChannelState::MonitorUpdateFailed as u32, ChannelState::MonitorUpdateFailed as u32); self.channel_state &= !(ChannelState::MonitorUpdateFailed as u32); @@ -3091,15 +3147,20 @@ impl Channel { }) } else { None }; - let mut forwards = Vec::new(); - mem::swap(&mut forwards, &mut self.monitor_pending_forwards); - let mut failures = Vec::new(); - mem::swap(&mut failures, &mut self.monitor_pending_failures); + let mut accepted_htlcs = Vec::new(); + mem::swap(&mut accepted_htlcs, &mut self.monitor_pending_forwards); + let mut failed_htlcs = Vec::new(); + mem::swap(&mut failed_htlcs, &mut self.monitor_pending_failures); + let mut finalized_claimed_htlcs = Vec::new(); + mem::swap(&mut finalized_claimed_htlcs, &mut self.monitor_pending_finalized_fulfills); if self.channel_state & (ChannelState::PeerDisconnected as u32) != 0 { self.monitor_pending_revoke_and_ack = false; self.monitor_pending_commitment_signed = false; - return (None, None, RAACommitmentOrder::RevokeAndACKFirst, forwards, failures, funding_broadcastable, funding_locked); + return MonitorRestoreUpdates { + raa: None, commitment_update: None, order: RAACommitmentOrder::RevokeAndACKFirst, + accepted_htlcs, failed_htlcs, finalized_claimed_htlcs, funding_broadcastable, funding_locked + }; } let raa = if self.monitor_pending_revoke_and_ack { @@ -3116,7 +3177,9 @@ impl Channel { log_bytes!(self.channel_id()), if funding_broadcastable.is_some() { "a funding broadcastable, " } else { "" }, if commitment_update.is_some() { "a" } else { "no" }, if raa.is_some() { "an" } else { "no" }, match order { RAACommitmentOrder::CommitmentFirst => "commitment", RAACommitmentOrder::RevokeAndACKFirst => "RAA"}); - (raa, commitment_update, order, forwards, failures, funding_broadcastable, funding_locked) + MonitorRestoreUpdates { + raa, commitment_update, order, accepted_htlcs, failed_htlcs, finalized_claimed_htlcs, funding_broadcastable, funding_locked + } } pub fn update_fee(&mut self, fee_estimator: &F, msg: &msgs::UpdateFee) -> Result<(), ChannelError> @@ -5176,6 +5239,7 @@ impl Writeable for Channel { (5, self.config, required), (7, self.shutdown_scriptpubkey, option), (9, self.target_closing_feerate_sats_per_kw, option), + (11, self.monitor_pending_finalized_fulfills, vec_type), }); Ok(()) @@ -5409,6 +5473,7 @@ impl<'a, Signer: Sign, K: Deref> ReadableArgs<&'a K> for Channel let mut announcement_sigs = None; let mut target_closing_feerate_sats_per_kw = None; + let mut monitor_pending_finalized_fulfills = Some(Vec::new()); read_tlv_fields!(reader, { (0, announcement_sigs, option), (1, minimum_depth, option), @@ -5416,6 +5481,7 @@ impl<'a, Signer: Sign, K: Deref> ReadableArgs<&'a K> for Channel (5, config, option), // Note that if none is provided we will *not* overwrite the existing one. (7, shutdown_scriptpubkey, option), (9, target_closing_feerate_sats_per_kw, option), + (11, monitor_pending_finalized_fulfills, vec_type), }); let mut secp_ctx = Secp256k1::new(); @@ -5451,6 +5517,7 @@ impl<'a, Signer: Sign, K: Deref> ReadableArgs<&'a K> for Channel monitor_pending_commitment_signed, monitor_pending_forwards, monitor_pending_failures, + monitor_pending_finalized_fulfills: monitor_pending_finalized_fulfills.unwrap(), pending_update_fee, holding_cell_update_fee, @@ -5700,6 +5767,7 @@ mod tests { session_priv: SecretKey::from_slice(&hex::decode("0fffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff").unwrap()[..]).unwrap(), first_hop_htlc_msat: 548, payment_id: PaymentId([42; 32]), + payment_secret: None, } }); diff --git a/lightning/src/ln/channelmanager.rs b/lightning/src/ln/channelmanager.rs index f3085fe2011..9e4708dcfd6 100644 --- a/lightning/src/ln/channelmanager.rs +++ b/lightning/src/ln/channelmanager.rs @@ -145,7 +145,7 @@ pub(super) enum HTLCForwardInfo { } /// Tracks the inbound corresponding to an outbound HTLC -#[derive(Clone, PartialEq)] +#[derive(Clone, Hash, PartialEq, Eq)] pub(crate) struct HTLCPreviousHopData { short_channel_id: u64, htlc_id: u64, @@ -189,7 +189,8 @@ impl Readable for PaymentId { } } /// Tracks the inbound corresponding to an outbound HTLC -#[derive(Clone, PartialEq)] +#[allow(clippy::derive_hash_xor_eq)] // Our Hash is faithful to the data, we just don't have SecretKey::hash +#[derive(Clone, PartialEq, Eq)] pub(crate) enum HTLCSource { PreviousHopData(HTLCPreviousHopData), OutboundRoute { @@ -199,8 +200,28 @@ pub(crate) enum HTLCSource { /// doing a double-pass on route when we get a failure back first_hop_htlc_msat: u64, payment_id: PaymentId, + payment_secret: Option, }, } +#[allow(clippy::derive_hash_xor_eq)] // Our Hash is faithful to the data, we just don't have SecretKey::hash +impl core::hash::Hash for HTLCSource { + fn hash(&self, hasher: &mut H) { + match self { + HTLCSource::PreviousHopData(prev_hop_data) => { + 0u8.hash(hasher); + prev_hop_data.hash(hasher); + }, + HTLCSource::OutboundRoute { path, session_priv, payment_id, payment_secret, first_hop_htlc_msat } => { + 1u8.hash(hasher); + path.hash(hasher); + session_priv[..].hash(hasher); + payment_id.hash(hasher); + payment_secret.hash(hasher); + first_hop_htlc_msat.hash(hasher); + }, + } + } +} #[cfg(test)] impl HTLCSource { pub fn dummy() -> Self { @@ -209,6 +230,7 @@ impl HTLCSource { session_priv: SecretKey::from_slice(&[1; 32]).unwrap(), first_hop_htlc_msat: 0, payment_id: PaymentId([2; 32]), + payment_secret: None, } } } @@ -416,19 +438,51 @@ pub(crate) enum PendingOutboundPayment { /// Our best known block height at the time this payment was initiated. starting_block_height: u32, }, + /// When a pending payment is fulfilled, we continue tracking it until all pending HTLCs have + /// been resolved. This ensures we don't look up pending payments in ChannelMonitors on restart + /// and add a pending payment that was already fulfilled. + Fulfilled { + session_privs: HashSet<[u8; 32]>, + }, } impl PendingOutboundPayment { - fn remove(&mut self, session_priv: &[u8; 32], part_amt_msat: u64) -> bool { + fn is_retryable(&self) -> bool { + match self { + PendingOutboundPayment::Retryable { .. } => true, + _ => false, + } + } + fn is_fulfilled(&self) -> bool { + match self { + PendingOutboundPayment::Fulfilled { .. } => true, + _ => false, + } + } + + fn mark_fulfilled(&mut self) { + let mut session_privs = HashSet::new(); + core::mem::swap(&mut session_privs, match self { + PendingOutboundPayment::Legacy { session_privs } | + PendingOutboundPayment::Retryable { session_privs, .. } | + PendingOutboundPayment::Fulfilled { session_privs } + => session_privs + }); + *self = PendingOutboundPayment::Fulfilled { session_privs }; + } + + /// panics if part_amt_msat is None and !self.is_fulfilled + fn remove(&mut self, session_priv: &[u8; 32], part_amt_msat: Option) -> bool { let remove_res = match self { PendingOutboundPayment::Legacy { session_privs } | - PendingOutboundPayment::Retryable { session_privs, .. } => { + PendingOutboundPayment::Retryable { session_privs, .. } | + PendingOutboundPayment::Fulfilled { session_privs } => { session_privs.remove(session_priv) } }; if remove_res { if let PendingOutboundPayment::Retryable { ref mut pending_amt_msat, .. } = self { - *pending_amt_msat -= part_amt_msat; + *pending_amt_msat -= part_amt_msat.expect("We must only not provide an amount if the payment was already fulfilled"); } } remove_res @@ -440,6 +494,7 @@ impl PendingOutboundPayment { PendingOutboundPayment::Retryable { session_privs, .. } => { session_privs.insert(session_priv) } + PendingOutboundPayment::Fulfilled { .. } => false }; if insert_res { if let PendingOutboundPayment::Retryable { ref mut pending_amt_msat, .. } = self { @@ -452,7 +507,8 @@ impl PendingOutboundPayment { fn remaining_parts(&self) -> usize { match self { PendingOutboundPayment::Legacy { session_privs } | - PendingOutboundPayment::Retryable { session_privs, .. } => { + PendingOutboundPayment::Retryable { session_privs, .. } | + PendingOutboundPayment::Fulfilled { session_privs } => { session_privs.len() } } @@ -1002,7 +1058,7 @@ macro_rules! handle_monitor_err { ($self: ident, $err: expr, $channel_state: expr, $entry: expr, $action_type: path, $resend_raa: expr, $resend_commitment: expr) => { handle_monitor_err!($self, $err, $channel_state, $entry, $action_type, $resend_raa, $resend_commitment, Vec::new(), Vec::new()) }; - ($self: ident, $err: expr, $short_to_id: expr, $chan: expr, $action_type: path, $resend_raa: expr, $resend_commitment: expr, $failed_forwards: expr, $failed_fails: expr, $chan_id: expr) => { + ($self: ident, $err: expr, $short_to_id: expr, $chan: expr, $action_type: path, $resend_raa: expr, $resend_commitment: expr, $failed_forwards: expr, $failed_fails: expr, $failed_finalized_fulfills: expr, $chan_id: expr) => { match $err { ChannelMonitorUpdateErr::PermanentFailure => { log_error!($self.logger, "Closing channel {} due to monitor update ChannelMonitorUpdateErr::PermanentFailure", log_bytes!($chan_id[..])); @@ -1023,7 +1079,7 @@ macro_rules! handle_monitor_err { (res, true) }, ChannelMonitorUpdateErr::TemporaryFailure => { - log_info!($self.logger, "Disabling channel {} due to monitor update TemporaryFailure. On restore will send {} and process {} forwards and {} fails", + log_info!($self.logger, "Disabling channel {} due to monitor update TemporaryFailure. On restore will send {} and process {} forwards, {} fails, and {} fulfill finalizations", log_bytes!($chan_id[..]), if $resend_commitment && $resend_raa { match $action_type { @@ -1034,25 +1090,29 @@ macro_rules! handle_monitor_err { else if $resend_raa { "RAA" } else { "nothing" }, (&$failed_forwards as &Vec<(PendingHTLCInfo, u64)>).len(), - (&$failed_fails as &Vec<(HTLCSource, PaymentHash, HTLCFailReason)>).len()); + (&$failed_fails as &Vec<(HTLCSource, PaymentHash, HTLCFailReason)>).len(), + (&$failed_finalized_fulfills as &Vec).len()); if !$resend_commitment { debug_assert!($action_type == RAACommitmentOrder::RevokeAndACKFirst || !$resend_raa); } if !$resend_raa { debug_assert!($action_type == RAACommitmentOrder::CommitmentFirst || !$resend_commitment); } - $chan.monitor_update_failed($resend_raa, $resend_commitment, $failed_forwards, $failed_fails); + $chan.monitor_update_failed($resend_raa, $resend_commitment, $failed_forwards, $failed_fails, $failed_finalized_fulfills); (Err(MsgHandleErrInternal::from_chan_no_close(ChannelError::Ignore("Failed to update ChannelMonitor".to_owned()), *$chan_id)), false) }, } }; - ($self: ident, $err: expr, $channel_state: expr, $entry: expr, $action_type: path, $resend_raa: expr, $resend_commitment: expr, $failed_forwards: expr, $failed_fails: expr) => { { - let (res, drop) = handle_monitor_err!($self, $err, $channel_state.short_to_id, $entry.get_mut(), $action_type, $resend_raa, $resend_commitment, $failed_forwards, $failed_fails, $entry.key()); + ($self: ident, $err: expr, $channel_state: expr, $entry: expr, $action_type: path, $resend_raa: expr, $resend_commitment: expr, $failed_forwards: expr, $failed_fails: expr, $failed_finalized_fulfills: expr) => { { + let (res, drop) = handle_monitor_err!($self, $err, $channel_state.short_to_id, $entry.get_mut(), $action_type, $resend_raa, $resend_commitment, $failed_forwards, $failed_fails, $failed_finalized_fulfills, $entry.key()); if drop { $entry.remove_entry(); } res } }; + ($self: ident, $err: expr, $channel_state: expr, $entry: expr, $action_type: path, $resend_raa: expr, $resend_commitment: expr, $failed_forwards: expr, $failed_fails: expr) => { + handle_monitor_err!($self, $err, $channel_state, $entry, $action_type, $resend_raa, $resend_commitment, $failed_forwards, $failed_fails, Vec::new()); + } } macro_rules! return_monitor_err { @@ -1441,7 +1501,7 @@ impl ChannelMana if let Some(monitor_update) = monitor_update { if let Err(e) = self.chain_monitor.update_channel(chan_entry.get().get_funding_txo().unwrap(), monitor_update) { let (result, is_permanent) = - handle_monitor_err!(self, e, channel_state.short_to_id, chan_entry.get_mut(), RAACommitmentOrder::CommitmentFirst, false, false, Vec::new(), Vec::new(), chan_entry.key()); + handle_monitor_err!(self, e, channel_state.short_to_id, chan_entry.get_mut(), RAACommitmentOrder::CommitmentFirst, false, false, Vec::new(), Vec::new(), Vec::new(), chan_entry.key()); if is_permanent { remove_channel!(channel_state, chan_entry); break result; @@ -1979,6 +2039,17 @@ impl ChannelMana let err: Result<(), _> = loop { let mut channel_lock = self.channel_state.lock().unwrap(); + + let mut pending_outbounds = self.pending_outbound_payments.lock().unwrap(); + let payment_entry = pending_outbounds.entry(payment_id); + if let hash_map::Entry::Occupied(payment) = &payment_entry { + if !payment.get().is_retryable() { + return Err(APIError::RouteError { + err: "Payment already completed" + }); + } + } + let id = match channel_lock.short_to_id.get(&path.first().unwrap().short_channel_id) { None => return Err(APIError::ChannelUnavailable{err: "No channel available with first hop!".to_owned()}), Some(id) => id.clone(), @@ -1999,11 +2070,11 @@ impl ChannelMana session_priv: session_priv.clone(), first_hop_htlc_msat: htlc_msat, payment_id, + payment_secret: payment_secret.clone(), }, onion_packet, &self.logger), channel_state, chan); - let mut pending_outbounds = self.pending_outbound_payments.lock().unwrap(); - let payment = pending_outbounds.entry(payment_id).or_insert_with(|| PendingOutboundPayment::Retryable { + let payment = payment_entry.or_insert_with(|| PendingOutboundPayment::Retryable { session_privs: HashSet::new(), pending_amt_msat: 0, payment_hash: *payment_hash, @@ -2199,7 +2270,12 @@ impl ChannelMana return Err(PaymentSendFailure::ParameterError(APIError::APIMisuseError { err: "Unable to retry payments that were initially sent on LDK versions prior to 0.0.102".to_string() })) - } + }, + PendingOutboundPayment::Fulfilled { .. } => { + return Err(PaymentSendFailure::ParameterError(APIError::RouteError { + err: "Payment already completed" + })); + }, } } else { return Err(PaymentSendFailure::ParameterError(APIError::APIMisuseError { @@ -2846,7 +2922,7 @@ impl ChannelMana let ret_err = match res { Ok(Some((update_fee, commitment_signed, monitor_update))) => { if let Err(e) = self.chain_monitor.update_channel(chan.get_funding_txo().unwrap(), monitor_update) { - let (res, drop) = handle_monitor_err!(self, e, short_to_id, chan, RAACommitmentOrder::CommitmentFirst, false, true, Vec::new(), Vec::new(), chan_id); + let (res, drop) = handle_monitor_err!(self, e, short_to_id, chan, RAACommitmentOrder::CommitmentFirst, false, true, Vec::new(), Vec::new(), Vec::new(), chan_id); if drop { retain_channel = false; } res } else { @@ -3027,7 +3103,9 @@ impl ChannelMana 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, path.last().unwrap().fee_msat) { + if payment.get_mut().remove(&session_priv_bytes, Some(path.last().unwrap().fee_msat)) && + !payment.get().is_fulfilled() + { self.pending_events.lock().unwrap().push( events::Event::PaymentPathFailed { payment_hash, @@ -3072,12 +3150,16 @@ impl ChannelMana session_priv_bytes.copy_from_slice(&session_priv[..]); let mut outbounds = self.pending_outbound_payments.lock().unwrap(); let mut all_paths_failed = false; - if let hash_map::Entry::Occupied(mut sessions) = outbounds.entry(payment_id) { - if !sessions.get_mut().remove(&session_priv_bytes, path.last().unwrap().fee_msat) { + if let hash_map::Entry::Occupied(mut payment) = outbounds.entry(payment_id) { + if !payment.get_mut().remove(&session_priv_bytes, Some(path.last().unwrap().fee_msat)) { log_trace!(self.logger, "Received duplicative fail for HTLC with payment_hash {}", log_bytes!(payment_hash.0)); return; } - if sessions.get().remaining_parts() == 0 { + if payment.get().is_fulfilled() { + log_trace!(self.logger, "Received failure of HTLC with payment_hash {} after payment completion", log_bytes!(payment_hash.0)); + return; + } + if payment.get().remaining_parts() == 0 { all_paths_failed = true; } } else { @@ -3325,6 +3407,23 @@ impl ChannelMana } else { unreachable!(); } } + fn finalize_claims(&self, mut sources: Vec) { + for source in sources.drain(..) { + if let HTLCSource::OutboundRoute { session_priv, payment_id, .. } = source { + 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) { + assert!(payment.get().is_fulfilled()); + payment.get_mut().remove(&session_priv_bytes, None); + if payment.get().remaining_parts() == 0 { + payment.remove(); + } + } + } + } + } + fn claim_funds_internal(&self, mut channel_state_lock: MutexGuard>, source: HTLCSource, payment_preimage: PaymentPreimage, forwarded_htlc_value_msat: Option, from_onchain: bool) { match source { HTLCSource::OutboundRoute { session_priv, payment_id, path, .. } => { @@ -3332,8 +3431,22 @@ impl ChannelMana let mut session_priv_bytes = [0; 32]; session_priv_bytes.copy_from_slice(&session_priv[..]); let mut outbounds = self.pending_outbound_payments.lock().unwrap(); - let found_payment = if let Some(mut sessions) = outbounds.remove(&payment_id) { - sessions.remove(&session_priv_bytes, path.last().unwrap().fee_msat) + let found_payment = if let hash_map::Entry::Occupied(mut payment) = outbounds.entry(payment_id) { + let found_payment = !payment.get().is_fulfilled(); + payment.get_mut().mark_fulfilled(); + if from_onchain { + // We currently immediately remove HTLCs which were fulfilled on-chain. + // This could potentially lead to removing a pending payment too early, + // with a reorg of one block causing us to re-add the fulfilled payment on + // restart. + // TODO: We should have a second monitor event that informs us of payments + // irrevocably fulfilled. + payment.get_mut().remove(&session_priv_bytes, Some(path.last().unwrap().fee_msat)); + if payment.get().remaining_parts() == 0 { + payment.remove(); + } + } + found_payment } else { false }; if found_payment { let payment_hash = PaymentHash(Sha256::hash(&payment_preimage.0).into_inner()); @@ -3408,7 +3521,7 @@ impl ChannelMana let _persistence_guard = PersistenceNotifierGuard::notify_on_drop(&self.total_consistency_lock, &self.persistence_notifier); let chan_restoration_res; - let mut pending_failures = { + let (mut pending_failures, finalized_claims) = { let mut channel_lock = self.channel_state.lock().unwrap(); let channel_state = &mut *channel_lock; let mut channel = match channel_state.by_id.entry(funding_txo.to_channel_id()) { @@ -3419,8 +3532,8 @@ impl ChannelMana return; } - let (raa, commitment_update, order, pending_forwards, pending_failures, funding_broadcastable, funding_locked) = channel.get_mut().monitor_updating_restored(&self.logger); - let channel_update = if funding_locked.is_some() && channel.get().is_usable() && !channel.get().should_announce() { + let updates = channel.get_mut().monitor_updating_restored(&self.logger); + let channel_update = if updates.funding_locked.is_some() && channel.get().is_usable() && !channel.get().should_announce() { // We only send a channel_update in the case where we are just now sending a // funding_locked and the channel is in a usable state. Further, we rely on the // normal announcement_signatures process to send a channel_update for public @@ -3430,13 +3543,14 @@ impl ChannelMana msg: self.get_channel_update_for_unicast(channel.get()).unwrap(), }) } else { None }; - chan_restoration_res = handle_chan_restoration_locked!(self, channel_lock, channel_state, channel, raa, commitment_update, order, None, pending_forwards, funding_broadcastable, funding_locked); + chan_restoration_res = handle_chan_restoration_locked!(self, channel_lock, channel_state, channel, updates.raa, updates.commitment_update, updates.order, None, updates.accepted_htlcs, updates.funding_broadcastable, updates.funding_locked); if let Some(upd) = channel_update { channel_state.pending_msg_events.push(upd); } - pending_failures + (updates.failed_htlcs, updates.finalized_claimed_htlcs) }; post_handle_chan_restoration!(self, chan_restoration_res); + self.finalize_claims(finalized_claims); for failure in pending_failures.drain(..) { self.fail_htlc_backwards_internal(self.channel_state.lock().unwrap(), failure.0, &failure.1, failure.2); } @@ -3525,7 +3639,7 @@ impl ChannelMana // hasn't persisted to disk yet - we can't lose money on a transaction that we haven't // accepted payment from yet. We do, however, need to wait to send our funding_locked // until we have persisted our monitor. - chan.monitor_update_failed(false, false, Vec::new(), Vec::new()); + chan.monitor_update_failed(false, false, Vec::new(), Vec::new(), Vec::new()); }, } } @@ -3643,7 +3757,7 @@ impl ChannelMana if let Some(monitor_update) = monitor_update { if let Err(e) = self.chain_monitor.update_channel(chan_entry.get().get_funding_txo().unwrap(), monitor_update) { let (result, is_permanent) = - handle_monitor_err!(self, e, channel_state.short_to_id, chan_entry.get_mut(), RAACommitmentOrder::CommitmentFirst, false, false, Vec::new(), Vec::new(), chan_entry.key()); + handle_monitor_err!(self, e, channel_state.short_to_id, chan_entry.get_mut(), RAACommitmentOrder::CommitmentFirst, false, false, Vec::new(), Vec::new(), Vec::new(), chan_entry.key()); if is_permanent { remove_channel!(channel_state, chan_entry); break result; @@ -3930,37 +4044,51 @@ impl ChannelMana break Err(MsgHandleErrInternal::send_err_msg_no_close("Got a message for a channel from the wrong node!".to_owned(), msg.channel_id)); } let was_frozen_for_monitor = chan.get().is_awaiting_monitor_update(); - let (commitment_update, pending_forwards, pending_failures, monitor_update, htlcs_to_fail_in) = - break_chan_entry!(self, chan.get_mut().revoke_and_ack(&msg, &self.logger), channel_state, chan); - htlcs_to_fail = htlcs_to_fail_in; - if let Err(e) = self.chain_monitor.update_channel(chan.get().get_funding_txo().unwrap(), monitor_update) { + let raa_updates = break_chan_entry!(self, + chan.get_mut().revoke_and_ack(&msg, &self.logger), channel_state, chan); + htlcs_to_fail = raa_updates.holding_cell_failed_htlcs; + if let Err(e) = self.chain_monitor.update_channel(chan.get().get_funding_txo().unwrap(), raa_updates.monitor_update) { if was_frozen_for_monitor { - assert!(commitment_update.is_none() && pending_forwards.is_empty() && pending_failures.is_empty()); + assert!(raa_updates.commitment_update.is_none()); + assert!(raa_updates.accepted_htlcs.is_empty()); + assert!(raa_updates.failed_htlcs.is_empty()); + assert!(raa_updates.finalized_claimed_htlcs.is_empty()); break Err(MsgHandleErrInternal::ignore_no_close("Previous monitor update failure prevented responses to RAA".to_owned())); } else { - if let Err(e) = handle_monitor_err!(self, e, channel_state, chan, RAACommitmentOrder::CommitmentFirst, false, commitment_update.is_some(), pending_forwards, pending_failures) { + if let Err(e) = handle_monitor_err!(self, e, channel_state, chan, + RAACommitmentOrder::CommitmentFirst, false, + raa_updates.commitment_update.is_some(), + raa_updates.accepted_htlcs, raa_updates.failed_htlcs, + raa_updates.finalized_claimed_htlcs) { break Err(e); } else { unreachable!(); } } } - if let Some(updates) = commitment_update { + if let Some(updates) = raa_updates.commitment_update { channel_state.pending_msg_events.push(events::MessageSendEvent::UpdateHTLCs { node_id: counterparty_node_id.clone(), updates, }); } - break Ok((pending_forwards, pending_failures, chan.get().get_short_channel_id().expect("RAA should only work on a short-id-available channel"), chan.get().get_funding_txo().unwrap())) + break Ok((raa_updates.accepted_htlcs, raa_updates.failed_htlcs, + raa_updates.finalized_claimed_htlcs, + chan.get().get_short_channel_id() + .expect("RAA should only work on a short-id-available channel"), + chan.get().get_funding_txo().unwrap())) }, hash_map::Entry::Vacant(_) => break Err(MsgHandleErrInternal::send_err_msg_no_close("Failed to find corresponding channel".to_owned(), msg.channel_id)) } }; self.fail_holding_cell_htlcs(htlcs_to_fail, msg.channel_id); match res { - Ok((pending_forwards, mut pending_failures, short_channel_id, channel_outpoint)) => { + Ok((pending_forwards, mut pending_failures, finalized_claim_htlcs, + short_channel_id, channel_outpoint)) => + { for failure in pending_failures.drain(..) { self.fail_htlc_backwards_internal(self.channel_state.lock().unwrap(), failure.0, &failure.1, failure.2); } self.forward_htlcs(&mut [(short_channel_id, channel_outpoint, pending_forwards)]); + self.finalize_claims(finalized_claim_htlcs); Ok(()) }, Err(e) => Err(e) @@ -4189,7 +4317,7 @@ impl ChannelMana if let Some((commitment_update, monitor_update)) = commitment_opt { if let Err(e) = self.chain_monitor.update_channel(chan.get_funding_txo().unwrap(), monitor_update) { has_monitor_update = true; - let (res, close_channel) = handle_monitor_err!(self, e, short_to_id, chan, RAACommitmentOrder::CommitmentFirst, false, true, Vec::new(), Vec::new(), channel_id); + let (res, close_channel) = handle_monitor_err!(self, e, short_to_id, chan, RAACommitmentOrder::CommitmentFirst, false, true, Vec::new(), Vec::new(), Vec::new(), channel_id); handle_errors.push((chan.get_counterparty_node_id(), res)); if close_channel { return false; } } else { @@ -5246,10 +5374,12 @@ impl Readable for HTLCSource { let mut first_hop_htlc_msat: u64 = 0; let mut path = Some(Vec::new()); let mut payment_id = None; + let mut payment_secret = None; read_tlv_fields!(reader, { (0, session_priv, required), (1, payment_id, option), (2, first_hop_htlc_msat, required), + (3, payment_secret, option), (4, path, vec_type), }); if payment_id.is_none() { @@ -5262,6 +5392,7 @@ impl Readable for HTLCSource { first_hop_htlc_msat: first_hop_htlc_msat, path: path.unwrap(), payment_id: payment_id.unwrap(), + payment_secret, }) } 1 => Ok(HTLCSource::PreviousHopData(Readable::read(reader)?)), @@ -5273,13 +5404,14 @@ impl Readable for HTLCSource { impl Writeable for HTLCSource { fn write(&self, writer: &mut W) -> Result<(), ::io::Error> { match self { - HTLCSource::OutboundRoute { ref session_priv, ref first_hop_htlc_msat, ref path, payment_id } => { + HTLCSource::OutboundRoute { ref session_priv, ref first_hop_htlc_msat, ref path, payment_id, payment_secret } => { 0u8.write(writer)?; let payment_id_opt = Some(payment_id); write_tlv_fields!(writer, { (0, session_priv, required), (1, payment_id_opt, option), (2, first_hop_htlc_msat, required), + (3, payment_secret, option), (4, path, vec_type), }); } @@ -5323,10 +5455,13 @@ impl_writeable_tlv_based!(PendingInboundPayment, { (8, min_value_msat, required), }); -impl_writeable_tlv_based_enum!(PendingOutboundPayment, +impl_writeable_tlv_based_enum_upgradable!(PendingOutboundPayment, (0, Legacy) => { (0, session_privs, required), }, + (1, Fulfilled) => { + (0, session_privs, required), + }, (2, Retryable) => { (0, session_privs, required), (2, payment_hash, required), @@ -5335,7 +5470,7 @@ impl_writeable_tlv_based_enum!(PendingOutboundPayment, (8, pending_amt_msat, required), (10, starting_block_height, required), }, -;); +); impl Writeable for ChannelManager where M::Target: chain::Watch, @@ -5428,7 +5563,9 @@ impl Writeable f // For backwards compat, write the session privs and their total length. let mut num_pending_outbounds_compat: u64 = 0; for (_, outbound) in pending_outbound_payments.iter() { - num_pending_outbounds_compat += outbound.remaining_parts() as u64; + if !outbound.is_fulfilled() { + num_pending_outbounds_compat += outbound.remaining_parts() as u64; + } } num_pending_outbounds_compat.write(writer)?; for (_, outbound) in pending_outbound_payments.iter() { @@ -5439,6 +5576,7 @@ impl Writeable f session_priv.write(writer)?; } } + PendingOutboundPayment::Fulfilled { .. } => {}, } } @@ -5449,7 +5587,8 @@ impl Writeable f PendingOutboundPayment::Legacy { session_privs } | PendingOutboundPayment::Retryable { session_privs, .. } => { pending_outbound_payments_no_retry.insert(*id, session_privs.clone()); - } + }, + _ => {}, } } write_tlv_fields!(writer, { @@ -5759,6 +5898,49 @@ impl<'a, Signer: Sign, M: Deref, T: Deref, K: Deref, F: Deref, L: Deref> outbounds.insert(id, PendingOutboundPayment::Legacy { session_privs }); } pending_outbound_payments = Some(outbounds); + } else { + // If we're tracking pending payments, ensure we haven't lost any by looking at the + // ChannelMonitor data for any channels for which we do not have authorative state + // (i.e. those for which we just force-closed above or we otherwise don't have a + // corresponding `Channel` at all). + // This avoids several edge-cases where we would otherwise "forget" about pending + // payments which are still in-flight via their on-chain state. + // We only rebuild the pending payments map if we were most recently serialized by + // 0.0.102+ + for (_, monitor) in args.channel_monitors { + if by_id.get(&monitor.get_funding_txo().0.to_channel_id()).is_none() { + for (htlc_source, htlc) in monitor.get_pending_outbound_htlcs() { + if let HTLCSource::OutboundRoute { payment_id, session_priv, path, payment_secret, .. } = htlc_source { + if path.is_empty() { + log_error!(args.logger, "Got an empty path for a pending payment"); + return Err(DecodeError::InvalidValue); + } + let path_amt = path.last().unwrap().fee_msat; + let mut session_priv_bytes = [0; 32]; + session_priv_bytes[..].copy_from_slice(&session_priv[..]); + match pending_outbound_payments.as_mut().unwrap().entry(payment_id) { + hash_map::Entry::Occupied(mut entry) => { + let newly_added = entry.get_mut().insert(session_priv_bytes, path_amt); + log_info!(args.logger, "{} a pending payment path for {} msat for session priv {} on an existing pending payment with payment hash {}", + if newly_added { "Added" } else { "Had" }, path_amt, log_bytes!(session_priv_bytes), log_bytes!(htlc.payment_hash.0)); + }, + hash_map::Entry::Vacant(entry) => { + entry.insert(PendingOutboundPayment::Retryable { + session_privs: [session_priv_bytes].iter().map(|a| *a).collect(), + payment_hash: htlc.payment_hash, + payment_secret, + pending_amt_msat: path_amt, + total_msat: path_amt, + starting_block_height: best_block_height, + }); + log_info!(args.logger, "Added a pending payment for {} msat with payment hash {} for path with session priv {}", + path_amt, log_bytes!(htlc.payment_hash.0), log_bytes!(session_priv_bytes)); + } + } + } + } + } + } } let mut secp_ctx = Secp256k1::new(); diff --git a/lightning/src/ln/functional_test_utils.rs b/lightning/src/ln/functional_test_utils.rs index 0f188ac38ee..fc57bafdd03 100644 --- a/lightning/src/ln/functional_test_utils.rs +++ b/lightning/src/ln/functional_test_utils.rs @@ -14,7 +14,7 @@ use chain::{BestBlock, Confirm, Listen, Watch}; use chain::channelmonitor::ChannelMonitor; use chain::transaction::OutPoint; use ln::{PaymentPreimage, PaymentHash, PaymentSecret}; -use ln::channelmanager::{ChainParameters, ChannelManager, ChannelManagerReadArgs, RAACommitmentOrder, PaymentSendFailure}; +use ln::channelmanager::{ChainParameters, ChannelManager, ChannelManagerReadArgs, RAACommitmentOrder, PaymentSendFailure, PaymentId}; use routing::network_graph::{NetGraphMsgHandler, NetworkGraph}; use routing::router::{Route, get_route}; use routing::scorer::Scorer; @@ -1156,10 +1156,11 @@ macro_rules! expect_payment_failed { } } -pub fn send_along_route_with_secret<'a, 'b, 'c>(origin_node: &Node<'a, 'b, 'c>, route: Route, expected_paths: &[&[&Node<'a, 'b, 'c>]], recv_value: u64, our_payment_hash: PaymentHash, our_payment_secret: PaymentSecret) { - origin_node.node.send_payment(&route, our_payment_hash, &Some(our_payment_secret)).unwrap(); +pub fn send_along_route_with_secret<'a, 'b, 'c>(origin_node: &Node<'a, 'b, 'c>, route: Route, expected_paths: &[&[&Node<'a, 'b, 'c>]], recv_value: u64, our_payment_hash: PaymentHash, our_payment_secret: PaymentSecret) -> PaymentId { + let payment_id = origin_node.node.send_payment(&route, our_payment_hash, &Some(our_payment_secret)).unwrap(); check_added_monitors!(origin_node, expected_paths.len()); pass_along_route(origin_node, expected_paths, recv_value, our_payment_hash, our_payment_secret); + payment_id } pub fn pass_along_path<'a, 'b, 'c>(origin_node: &Node<'a, 'b, 'c>, expected_path: &[&Node<'a, 'b, 'c>], recv_value: u64, our_payment_hash: PaymentHash, our_payment_secret: Option, ev: MessageSendEvent, payment_received_expected: bool, expected_preimage: Option) { @@ -1222,10 +1223,10 @@ pub fn pass_along_route<'a, 'b, 'c>(origin_node: &Node<'a, 'b, 'c>, expected_rou } } -pub fn send_along_route<'a, 'b, 'c>(origin_node: &Node<'a, 'b, 'c>, route: Route, expected_route: &[&Node<'a, 'b, 'c>], recv_value: u64) -> (PaymentPreimage, PaymentHash, PaymentSecret) { +pub fn send_along_route<'a, 'b, 'c>(origin_node: &Node<'a, 'b, 'c>, route: Route, expected_route: &[&Node<'a, 'b, 'c>], recv_value: u64) -> (PaymentPreimage, PaymentHash, PaymentSecret, PaymentId) { let (our_payment_preimage, our_payment_hash, our_payment_secret) = get_payment_preimage_hash!(expected_route.last().unwrap()); - send_along_route_with_secret(origin_node, route, &[expected_route], recv_value, our_payment_hash, our_payment_secret); - (our_payment_preimage, our_payment_hash, our_payment_secret) + let payment_id = send_along_route_with_secret(origin_node, route, &[expected_route], recv_value, our_payment_hash, our_payment_secret); + (our_payment_preimage, our_payment_hash, our_payment_secret, payment_id) } pub fn claim_payment_along_route<'a, 'b, 'c>(origin_node: &Node<'a, 'b, 'c>, expected_paths: &[&[&Node<'a, 'b, 'c>]], skip_last: bool, our_payment_preimage: PaymentPreimage) { @@ -1339,7 +1340,8 @@ pub fn route_payment<'a, 'b, 'c>(origin_node: &Node<'a, 'b, 'c>, expected_route: assert_eq!(hop.pubkey, node.node.get_our_node_id()); } - send_along_route(origin_node, route, expected_route, recv_value) + let res = send_along_route(origin_node, route, expected_route, recv_value); + (res.0, res.1, res.2) } pub fn route_over_limit<'a, 'b, 'c>(origin_node: &Node<'a, 'b, 'c>, expected_route: &[&Node<'a, 'b, 'c>], recv_value: u64) { diff --git a/lightning/src/ln/functional_tests.rs b/lightning/src/ln/functional_tests.rs index eb7d868a95d..4e2e1550209 100644 --- a/lightning/src/ln/functional_tests.rs +++ b/lightning/src/ln/functional_tests.rs @@ -12,7 +12,7 @@ //! claim outputs on-chain. use chain; -use chain::{Confirm, Listen, Watch, ChannelMonitorUpdateErr}; +use chain::{Confirm, Listen, Watch}; use chain::channelmonitor; use chain::channelmonitor::{ChannelMonitor, CLTV_CLAIM_BUFFER, LATENCY_GRACE_PERIOD_BLOCKS, ANTI_REORG_DELAY}; use chain::transaction::OutPoint; @@ -3689,7 +3689,7 @@ fn test_funding_peer_disconnect() { nodes[0].net_graph_msg_handler.handle_channel_update(&as_update).unwrap(); let (route, _, _, _) = get_route_and_payment_hash!(nodes[0], nodes[1], 1000000); - let (payment_preimage, _, _) = send_along_route(&nodes[0], route, &[&nodes[1]], 1000000); + let payment_preimage = send_along_route(&nodes[0], route, &[&nodes[1]], 1000000).0; claim_payment(&nodes[0], &[&nodes[1]], payment_preimage); // Check that after deserialization and reconnection we can still generate an identical @@ -4100,145 +4100,6 @@ fn test_no_txn_manager_serialize_deserialize() { send_payment(&nodes[0], &[&nodes[1]], 1000000); } -fn do_test_dup_htlc_onchain_fails_on_reload(persist_manager_post_event: bool) { - // When a Channel is closed, any outbound HTLCs which were relayed through it are simply - // dropped when the Channel is. From there, the ChannelManager relies on the ChannelMonitor - // having a copy of the relevant fail-/claim-back data and processes the HTLC fail/claim when - // the ChannelMonitor tells it to. - // - // If, due to an on-chain event, an HTLC is failed/claimed, we should avoid providing the - // ChannelManager the HTLC event until after the monitor is re-persisted. This should prevent a - // duplicate HTLC fail/claim (e.g. via a PaymentPathFailed event). - let chanmon_cfgs = create_chanmon_cfgs(2); - let node_cfgs = create_node_cfgs(2, &chanmon_cfgs); - let node_chanmgrs = create_node_chanmgrs(2, &node_cfgs, &[None, None]); - let persister: test_utils::TestPersister; - let new_chain_monitor: test_utils::TestChainMonitor; - let nodes_0_deserialized: ChannelManager; - let mut nodes = create_network(2, &node_cfgs, &node_chanmgrs); - - let (_, _, chan_id, funding_tx) = create_announced_chan_between_nodes(&nodes, 0, 1, InitFeatures::known(), InitFeatures::known()); - - // Route a payment, but force-close the channel before the HTLC fulfill message arrives at - // nodes[0]. - let (payment_preimage, _, _) = route_payment(&nodes[0], &[&nodes[1]], 10000000); - nodes[0].node.force_close_channel(&nodes[0].node.list_channels()[0].channel_id).unwrap(); - check_closed_broadcast!(nodes[0], true); - check_added_monitors!(nodes[0], 1); - check_closed_event!(nodes[0], 1, ClosureReason::HolderForceClosed); - - nodes[0].node.peer_disconnected(&nodes[1].node.get_our_node_id(), false); - nodes[1].node.peer_disconnected(&nodes[0].node.get_our_node_id(), false); - - // Connect blocks until the CLTV timeout is up so that we get an HTLC-Timeout transaction - connect_blocks(&nodes[0], TEST_FINAL_CLTV + LATENCY_GRACE_PERIOD_BLOCKS + 1); - let node_txn = nodes[0].tx_broadcaster.txn_broadcasted.lock().unwrap().split_off(0); - assert_eq!(node_txn.len(), 3); - assert_eq!(node_txn[0], node_txn[1]); - check_spends!(node_txn[1], funding_tx); - check_spends!(node_txn[2], node_txn[1]); - - assert!(nodes[1].node.claim_funds(payment_preimage)); - check_added_monitors!(nodes[1], 1); - - let mut header = BlockHeader { version: 0x20000000, prev_blockhash: nodes[1].best_block_hash(), merkle_root: Default::default(), time: 42, bits: 42, nonce: 42 }; - connect_block(&nodes[1], &Block { header, txdata: vec![node_txn[1].clone()]}); - check_closed_broadcast!(nodes[1], true); - check_added_monitors!(nodes[1], 1); - check_closed_event!(nodes[1], 1, ClosureReason::CommitmentTxConfirmed); - let claim_txn = nodes[1].tx_broadcaster.txn_broadcasted.lock().unwrap().split_off(0); - - header.prev_blockhash = nodes[0].best_block_hash(); - connect_block(&nodes[0], &Block { header, txdata: vec![node_txn[1].clone()]}); - - // Now connect the HTLC claim transaction with the ChainMonitor-generated ChannelMonitor update - // returning TemporaryFailure. This should cause the claim event to never make its way to the - // ChannelManager. - chanmon_cfgs[0].persister.chain_sync_monitor_persistences.lock().unwrap().clear(); - chanmon_cfgs[0].persister.set_update_ret(Err(ChannelMonitorUpdateErr::TemporaryFailure)); - - header.prev_blockhash = nodes[0].best_block_hash(); - let claim_block = Block { header, txdata: claim_txn }; - connect_block(&nodes[0], &claim_block); - - let funding_txo = OutPoint { txid: funding_tx.txid(), index: 0 }; - let mon_updates: Vec<_> = chanmon_cfgs[0].persister.chain_sync_monitor_persistences.lock().unwrap() - .get_mut(&funding_txo).unwrap().drain().collect(); - assert_eq!(mon_updates.len(), 1); - assert!(nodes[0].chain_monitor.release_pending_monitor_events().is_empty()); - assert!(nodes[0].node.get_and_clear_pending_events().is_empty()); - - // If we persist the ChannelManager here, we should get the PaymentSent event after - // deserialization. - let mut chan_manager_serialized = test_utils::TestVecWriter(Vec::new()); - if !persist_manager_post_event { - nodes[0].node.write(&mut chan_manager_serialized).unwrap(); - } - - // Now persist the ChannelMonitor and inform the ChainMonitor that we're done, generating the - // payment sent event. - chanmon_cfgs[0].persister.set_update_ret(Ok(())); - let mut chan_0_monitor_serialized = test_utils::TestVecWriter(Vec::new()); - get_monitor!(nodes[0], chan_id).write(&mut chan_0_monitor_serialized).unwrap(); - nodes[0].chain_monitor.chain_monitor.channel_monitor_updated(funding_txo, mon_updates[0]).unwrap(); - expect_payment_sent!(nodes[0], payment_preimage); - - // If we persist the ChannelManager after we get the PaymentSent event, we shouldn't get it - // twice. - if persist_manager_post_event { - nodes[0].node.write(&mut chan_manager_serialized).unwrap(); - } - - // Now reload nodes[0]... - persister = test_utils::TestPersister::new(); - let keys_manager = &chanmon_cfgs[0].keys_manager; - new_chain_monitor = test_utils::TestChainMonitor::new(Some(nodes[0].chain_source), nodes[0].tx_broadcaster.clone(), nodes[0].logger, node_cfgs[0].fee_estimator, &persister, keys_manager); - nodes[0].chain_monitor = &new_chain_monitor; - let mut chan_0_monitor_read = &chan_0_monitor_serialized.0[..]; - let (_, mut chan_0_monitor) = <(BlockHash, ChannelMonitor)>::read( - &mut chan_0_monitor_read, keys_manager).unwrap(); - assert!(chan_0_monitor_read.is_empty()); - - let (_, nodes_0_deserialized_tmp) = { - let mut channel_monitors = HashMap::new(); - channel_monitors.insert(chan_0_monitor.get_funding_txo().0, &mut chan_0_monitor); - <(BlockHash, ChannelManager)> - ::read(&mut io::Cursor::new(&chan_manager_serialized.0[..]), ChannelManagerReadArgs { - default_config: Default::default(), - keys_manager, - fee_estimator: node_cfgs[0].fee_estimator, - chain_monitor: nodes[0].chain_monitor, - tx_broadcaster: nodes[0].tx_broadcaster.clone(), - logger: nodes[0].logger, - channel_monitors, - }).unwrap() - }; - nodes_0_deserialized = nodes_0_deserialized_tmp; - - assert!(nodes[0].chain_monitor.watch_channel(chan_0_monitor.get_funding_txo().0, chan_0_monitor).is_ok()); - check_added_monitors!(nodes[0], 1); - nodes[0].node = &nodes_0_deserialized; - - if persist_manager_post_event { - assert!(nodes[0].node.get_and_clear_pending_events().is_empty()); - } else { - expect_payment_sent!(nodes[0], payment_preimage); - } - - // Note that if we re-connect the block which exposed nodes[0] to the payment preimage (but - // which the current ChannelMonitor has not seen), the ChannelManager's de-duplication of - // payment events should kick in, leaving us with no pending events here. - let height = nodes[0].blocks.lock().unwrap().len() as u32 - 1; - nodes[0].chain_monitor.chain_monitor.block_connected(&claim_block, height); - assert!(nodes[0].node.get_and_clear_pending_events().is_empty()); -} - -#[test] -fn test_dup_htlc_onchain_fails_on_reload() { - do_test_dup_htlc_onchain_fails_on_reload(true); - do_test_dup_htlc_onchain_fails_on_reload(false); -} - #[test] fn test_manager_serialize_deserialize_events() { // This test makes sure the events field in ChannelManager survives de/serialization diff --git a/lightning/src/ln/onion_route_tests.rs b/lightning/src/ln/onion_route_tests.rs index 163c4ae15d5..c6f55e384d8 100644 --- a/lightning/src/ln/onion_route_tests.rs +++ b/lightning/src/ln/onion_route_tests.rs @@ -479,7 +479,7 @@ fn test_onion_failure() { // Test a positive test-case with one extra msat, meeting the minimum. bogus_route.paths[0][route_len-1].fee_msat = amt_to_forward + 1; - let (preimage, _, _) = send_along_route(&nodes[0], bogus_route, &[&nodes[1], &nodes[2]], amt_to_forward+1); + let preimage = send_along_route(&nodes[0], bogus_route, &[&nodes[1], &nodes[2]], amt_to_forward+1).0; claim_payment(&nodes[0], &[&nodes[1], &nodes[2]], preimage); //TODO: with new config API, we will be able to generate both valid and diff --git a/lightning/src/ln/payment_tests.rs b/lightning/src/ln/payment_tests.rs index c622309a4b5..1b9836621e7 100644 --- a/lightning/src/ln/payment_tests.rs +++ b/lightning/src/ln/payment_tests.rs @@ -11,16 +11,24 @@ //! serialization ordering between ChannelManager/ChannelMonitors and ensuring we can still retry //! payments thereafter. +use chain::{ChannelMonitorUpdateErr, Confirm, Listen, Watch}; +use chain::channelmonitor::{ANTI_REORG_DELAY, ChannelMonitor, LATENCY_GRACE_PERIOD_BLOCKS}; +use chain::transaction::OutPoint; use ln::{PaymentPreimage, PaymentHash}; -use ln::channelmanager::{PaymentId, PaymentSendFailure}; +use ln::channelmanager::{BREAKDOWN_TIMEOUT, ChannelManager, ChannelManagerReadArgs, PaymentId, PaymentSendFailure}; use ln::features::InitFeatures; use ln::msgs; -use ln::msgs::ChannelMessageHandler; -use util::events::{Event, MessageSendEvent, MessageSendEventsProvider}; +use ln::msgs::{ChannelMessageHandler, ErrorAction}; +use util::events::{ClosureReason, Event, MessageSendEvent, MessageSendEventsProvider}; +use util::test_utils; use util::errors::APIError; +use util::enforcing_trait_impls::EnforcingSigner; +use util::ser::{ReadableArgs, Writeable}; +use io; use bitcoin::hashes::sha256::Hash as Sha256; use bitcoin::hashes::Hash; +use bitcoin::{Block, BlockHeader, BlockHash}; use prelude::*; @@ -266,3 +274,430 @@ fn no_pending_leak_on_initial_send_failure() { assert!(!nodes[0].node.has_pending_payments()); } + +fn do_retry_with_no_persist(confirm_before_reload: bool) { + // If we send a pending payment and `send_payment` returns success, we should always either + // return a payment failure event or a payment success event, and on failure the payment should + // be retryable. + // + // In order to do so when the ChannelManager isn't immediately persisted (which is normal - its + // always persisted asynchronously), the ChannelManager has to reload some payment data from + // ChannelMonitor(s) in some cases. This tests that reloading. + // + // `confirm_before_reload` confirms the channel-closing commitment transaction on-chain prior + // to reloading the ChannelManager, increasing test coverage in ChannelMonitor HTLC tracking + // which has separate codepaths for "commitment transaction already confirmed" and not. + let chanmon_cfgs = create_chanmon_cfgs(3); + let node_cfgs = create_node_cfgs(3, &chanmon_cfgs); + let node_chanmgrs = create_node_chanmgrs(3, &node_cfgs, &[None, None, None]); + let persister: test_utils::TestPersister; + let new_chain_monitor: test_utils::TestChainMonitor; + let nodes_0_deserialized: ChannelManager; + let mut nodes = create_network(3, &node_cfgs, &node_chanmgrs); + + let (_, _, chan_id, funding_tx) = create_announced_chan_between_nodes(&nodes, 0, 1, InitFeatures::known(), InitFeatures::known()); + create_announced_chan_between_nodes(&nodes, 1, 2, InitFeatures::known(), InitFeatures::known()); + + // Serialize the ChannelManager prior to sending payments + let nodes_0_serialized = nodes[0].node.encode(); + + // Send two payments - one which will get to nodes[2] and will be claimed, one which we'll time + // out and retry. + let (route, payment_hash, payment_preimage, payment_secret) = get_route_and_payment_hash!(nodes[0], nodes[2], 1_000_000); + let (payment_preimage_1, _, _, payment_id_1) = send_along_route(&nodes[0], route.clone(), &[&nodes[1], &nodes[2]], 1_000_000); + let payment_id = nodes[0].node.send_payment(&route, payment_hash, &Some(payment_secret)).unwrap(); + check_added_monitors!(nodes[0], 1); + + let mut events = nodes[0].node.get_and_clear_pending_msg_events(); + assert_eq!(events.len(), 1); + let payment_event = SendEvent::from_event(events.pop().unwrap()); + assert_eq!(payment_event.node_id, nodes[1].node.get_our_node_id()); + + // We relay the payment to nodes[1] while its disconnected from nodes[2], causing the payment + // to be returned immediately to nodes[0], without having nodes[2] fail the inbound payment + // which would prevent retry. + nodes[1].node.peer_disconnected(&nodes[2].node.get_our_node_id(), false); + nodes[2].node.peer_disconnected(&nodes[1].node.get_our_node_id(), false); + + nodes[1].node.handle_update_add_htlc(&nodes[0].node.get_our_node_id(), &payment_event.msgs[0]); + commitment_signed_dance!(nodes[1], nodes[0], payment_event.commitment_msg, false, true); + // nodes[1] now immediately fails the HTLC as the next-hop channel is disconnected + let _ = get_htlc_update_msgs!(nodes[1], nodes[0].node.get_our_node_id()); + + reconnect_nodes(&nodes[1], &nodes[2], (false, false), (0, 0), (0, 0), (0, 0), (0, 0), (0, 0), (false, false)); + + let as_commitment_tx = get_local_commitment_txn!(nodes[0], chan_id)[0].clone(); + if confirm_before_reload { + mine_transaction(&nodes[0], &as_commitment_tx); + nodes[0].tx_broadcaster.txn_broadcasted.lock().unwrap().clear(); + } + + // The ChannelMonitor should always be the latest version, as we're required to persist it + // during the `commitment_signed_dance!()`. + let mut chan_0_monitor_serialized = test_utils::TestVecWriter(Vec::new()); + get_monitor!(nodes[0], chan_id).write(&mut chan_0_monitor_serialized).unwrap(); + + persister = test_utils::TestPersister::new(); + let keys_manager = &chanmon_cfgs[0].keys_manager; + new_chain_monitor = test_utils::TestChainMonitor::new(Some(nodes[0].chain_source), nodes[0].tx_broadcaster.clone(), nodes[0].logger, node_cfgs[0].fee_estimator, &persister, keys_manager); + nodes[0].chain_monitor = &new_chain_monitor; + let mut chan_0_monitor_read = &chan_0_monitor_serialized.0[..]; + let (_, mut chan_0_monitor) = <(BlockHash, ChannelMonitor)>::read( + &mut chan_0_monitor_read, keys_manager).unwrap(); + assert!(chan_0_monitor_read.is_empty()); + + let mut nodes_0_read = &nodes_0_serialized[..]; + let (_, nodes_0_deserialized_tmp) = { + let mut channel_monitors = HashMap::new(); + channel_monitors.insert(chan_0_monitor.get_funding_txo().0, &mut chan_0_monitor); + <(BlockHash, ChannelManager)>::read(&mut nodes_0_read, ChannelManagerReadArgs { + default_config: test_default_channel_config(), + keys_manager, + fee_estimator: node_cfgs[0].fee_estimator, + chain_monitor: nodes[0].chain_monitor, + tx_broadcaster: nodes[0].tx_broadcaster.clone(), + logger: nodes[0].logger, + channel_monitors, + }).unwrap() + }; + nodes_0_deserialized = nodes_0_deserialized_tmp; + assert!(nodes_0_read.is_empty()); + + assert!(nodes[0].chain_monitor.watch_channel(chan_0_monitor.get_funding_txo().0, chan_0_monitor).is_ok()); + nodes[0].node = &nodes_0_deserialized; + check_added_monitors!(nodes[0], 1); + + // On reload, the ChannelManager should realize it is stale compared to the ChannelMonitor and + // force-close the channel. + check_closed_event!(nodes[0], 1, ClosureReason::OutdatedChannelManager); + assert!(nodes[0].node.list_channels().is_empty()); + assert!(nodes[0].node.has_pending_payments()); + let as_broadcasted_txn = nodes[0].tx_broadcaster.txn_broadcasted.lock().unwrap().split_off(0); + assert_eq!(as_broadcasted_txn.len(), 1); + assert_eq!(as_broadcasted_txn[0], as_commitment_tx); + + nodes[1].node.peer_disconnected(&nodes[0].node.get_our_node_id(), false); + nodes[0].node.peer_connected(&nodes[1].node.get_our_node_id(), &msgs::Init { features: InitFeatures::known()}); + assert!(nodes[0].node.get_and_clear_pending_msg_events().is_empty()); + + // Now nodes[1] should send a channel reestablish, which nodes[0] will respond to with an + // error, as the channel has hit the chain. + nodes[1].node.peer_connected(&nodes[0].node.get_our_node_id(), &msgs::Init { features: InitFeatures::known()}); + let bs_reestablish = get_event_msg!(nodes[1], MessageSendEvent::SendChannelReestablish, nodes[0].node.get_our_node_id()); + nodes[0].node.handle_channel_reestablish(&nodes[1].node.get_our_node_id(), &bs_reestablish); + let as_err = nodes[0].node.get_and_clear_pending_msg_events(); + assert_eq!(as_err.len(), 1); + match as_err[0] { + MessageSendEvent::HandleError { node_id, action: msgs::ErrorAction::SendErrorMessage { ref msg } } => { + assert_eq!(node_id, nodes[1].node.get_our_node_id()); + nodes[1].node.handle_error(&nodes[0].node.get_our_node_id(), msg); + check_closed_event!(nodes[1], 1, ClosureReason::CounterpartyForceClosed { peer_msg: "Failed to find corresponding channel".to_string() }); + check_added_monitors!(nodes[1], 1); + assert_eq!(nodes[1].tx_broadcaster.txn_broadcasted.lock().unwrap().split_off(0).len(), 1); + }, + _ => panic!("Unexpected event"), + } + check_closed_broadcast!(nodes[1], false); + + // Now claim the first payment, which should allow nodes[1] to claim the payment on-chain when + // we close in a moment. + nodes[2].node.claim_funds(payment_preimage_1); + check_added_monitors!(nodes[2], 1); + let htlc_fulfill_updates = get_htlc_update_msgs!(nodes[2], nodes[1].node.get_our_node_id()); + nodes[1].node.handle_update_fulfill_htlc(&nodes[2].node.get_our_node_id(), &htlc_fulfill_updates.update_fulfill_htlcs[0]); + check_added_monitors!(nodes[1], 1); + commitment_signed_dance!(nodes[1], nodes[2], htlc_fulfill_updates.commitment_signed, false); + + if confirm_before_reload { + let best_block = nodes[0].blocks.lock().unwrap().last().unwrap().clone(); + nodes[0].node.best_block_updated(&best_block.0, best_block.1); + } + + // Create a new channel on which to retry the payment before we fail the payment via the + // HTLC-Timeout transaction. This avoids ChannelManager timing out the payment due to us + // connecting several blocks while creating the channel (implying time has passed). + create_announced_chan_between_nodes(&nodes, 0, 1, InitFeatures::known(), InitFeatures::known()); + assert_eq!(nodes[0].node.list_usable_channels().len(), 1); + + mine_transaction(&nodes[1], &as_commitment_tx); + let bs_htlc_claim_txn = nodes[1].tx_broadcaster.txn_broadcasted.lock().unwrap().split_off(0); + assert_eq!(bs_htlc_claim_txn.len(), 1); + check_spends!(bs_htlc_claim_txn[0], as_commitment_tx); + expect_payment_forwarded!(nodes[1], None, false); + + mine_transaction(&nodes[0], &as_commitment_tx); + mine_transaction(&nodes[0], &bs_htlc_claim_txn[0]); + expect_payment_sent!(nodes[0], payment_preimage_1); + connect_blocks(&nodes[0], TEST_FINAL_CLTV*4 + 20); + let as_htlc_timeout_txn = nodes[0].tx_broadcaster.txn_broadcasted.lock().unwrap().split_off(0); + check_spends!(as_htlc_timeout_txn[2], funding_tx); + check_spends!(as_htlc_timeout_txn[0], as_commitment_tx); + check_spends!(as_htlc_timeout_txn[1], as_commitment_tx); + assert_eq!(as_htlc_timeout_txn.len(), 3); + if as_htlc_timeout_txn[0].input[0].previous_output == bs_htlc_claim_txn[0].input[0].previous_output { + confirm_transaction(&nodes[0], &as_htlc_timeout_txn[1]); + } else { + confirm_transaction(&nodes[0], &as_htlc_timeout_txn[0]); + } + nodes[0].tx_broadcaster.txn_broadcasted.lock().unwrap().clear(); + expect_payment_failed!(nodes[0], payment_hash, false); + + // Finally, retry the payment (which was reloaded from the ChannelMonitor when nodes[0] was + // reloaded) via a route over the new channel, which work without issue and eventually be + // received and claimed at the recipient just like any other payment. + let (new_route, _, _, _) = get_route_and_payment_hash!(nodes[0], nodes[2], 1_000_000); + + assert!(nodes[0].node.retry_payment(&new_route, payment_id_1).is_err()); // Shouldn't be allowed to retry a fulfilled payment + nodes[0].node.retry_payment(&new_route, payment_id).unwrap(); + check_added_monitors!(nodes[0], 1); + let mut events = nodes[0].node.get_and_clear_pending_msg_events(); + assert_eq!(events.len(), 1); + pass_along_path(&nodes[0], &[&nodes[1], &nodes[2]], 1_000_000, payment_hash, Some(payment_secret), events.pop().unwrap(), true, None); + claim_payment_along_route(&nodes[0], &[&[&nodes[1], &nodes[2]]], false, payment_preimage); +} + +#[test] +fn retry_with_no_persist() { + do_retry_with_no_persist(true); + do_retry_with_no_persist(false); +} + +fn do_test_dup_htlc_onchain_fails_on_reload(persist_manager_post_event: bool, confirm_commitment_tx: bool, payment_timeout: bool) { + // When a Channel is closed, any outbound HTLCs which were relayed through it are simply + // dropped when the Channel is. From there, the ChannelManager relies on the ChannelMonitor + // having a copy of the relevant fail-/claim-back data and processes the HTLC fail/claim when + // the ChannelMonitor tells it to. + // + // If, due to an on-chain event, an HTLC is failed/claimed, we should avoid providing the + // ChannelManager the HTLC event until after the monitor is re-persisted. This should prevent a + // duplicate HTLC fail/claim (e.g. via a PaymentPathFailed event). + let chanmon_cfgs = create_chanmon_cfgs(2); + let node_cfgs = create_node_cfgs(2, &chanmon_cfgs); + let node_chanmgrs = create_node_chanmgrs(2, &node_cfgs, &[None, None]); + let persister: test_utils::TestPersister; + let new_chain_monitor: test_utils::TestChainMonitor; + let nodes_0_deserialized: ChannelManager; + let mut nodes = create_network(2, &node_cfgs, &node_chanmgrs); + + let (_, _, chan_id, funding_tx) = create_announced_chan_between_nodes(&nodes, 0, 1, InitFeatures::known(), InitFeatures::known()); + + // Route a payment, but force-close the channel before the HTLC fulfill message arrives at + // nodes[0]. + let (payment_preimage, payment_hash, _) = route_payment(&nodes[0], &[&nodes[1]], 10000000); + nodes[0].node.force_close_channel(&nodes[0].node.list_channels()[0].channel_id).unwrap(); + check_closed_broadcast!(nodes[0], true); + check_added_monitors!(nodes[0], 1); + check_closed_event!(nodes[0], 1, ClosureReason::HolderForceClosed); + + nodes[0].node.peer_disconnected(&nodes[1].node.get_our_node_id(), false); + nodes[1].node.peer_disconnected(&nodes[0].node.get_our_node_id(), false); + + // Connect blocks until the CLTV timeout is up so that we get an HTLC-Timeout transaction + connect_blocks(&nodes[0], TEST_FINAL_CLTV + LATENCY_GRACE_PERIOD_BLOCKS + 1); + let node_txn = nodes[0].tx_broadcaster.txn_broadcasted.lock().unwrap().split_off(0); + assert_eq!(node_txn.len(), 3); + assert_eq!(node_txn[0], node_txn[1]); + check_spends!(node_txn[1], funding_tx); + check_spends!(node_txn[2], node_txn[1]); + let timeout_txn = vec![node_txn[2].clone()]; + + assert!(nodes[1].node.claim_funds(payment_preimage)); + check_added_monitors!(nodes[1], 1); + + let mut header = BlockHeader { version: 0x20000000, prev_blockhash: nodes[1].best_block_hash(), merkle_root: Default::default(), time: 42, bits: 42, nonce: 42 }; + connect_block(&nodes[1], &Block { header, txdata: vec![node_txn[1].clone()]}); + check_closed_broadcast!(nodes[1], true); + check_added_monitors!(nodes[1], 1); + check_closed_event!(nodes[1], 1, ClosureReason::CommitmentTxConfirmed); + let claim_txn = nodes[1].tx_broadcaster.txn_broadcasted.lock().unwrap().split_off(0); + + header.prev_blockhash = nodes[0].best_block_hash(); + connect_block(&nodes[0], &Block { header, txdata: vec![node_txn[1].clone()]}); + + if confirm_commitment_tx { + connect_blocks(&nodes[0], BREAKDOWN_TIMEOUT as u32 - 1); + } + + header.prev_blockhash = nodes[0].best_block_hash(); + let claim_block = Block { header, txdata: if payment_timeout { timeout_txn } else { claim_txn } }; + + if payment_timeout { + assert!(confirm_commitment_tx); // Otherwise we're spending below our CSV! + connect_block(&nodes[0], &claim_block); + connect_blocks(&nodes[0], ANTI_REORG_DELAY - 2); + } + + // Now connect the HTLC claim transaction with the ChainMonitor-generated ChannelMonitor update + // returning TemporaryFailure. This should cause the claim event to never make its way to the + // ChannelManager. + chanmon_cfgs[0].persister.chain_sync_monitor_persistences.lock().unwrap().clear(); + chanmon_cfgs[0].persister.set_update_ret(Err(ChannelMonitorUpdateErr::TemporaryFailure)); + + if payment_timeout { + connect_blocks(&nodes[0], 1); + } else { + connect_block(&nodes[0], &claim_block); + } + + let funding_txo = OutPoint { txid: funding_tx.txid(), index: 0 }; + let mon_updates: Vec<_> = chanmon_cfgs[0].persister.chain_sync_monitor_persistences.lock().unwrap() + .get_mut(&funding_txo).unwrap().drain().collect(); + assert_eq!(mon_updates.len(), 1); + assert!(nodes[0].chain_monitor.release_pending_monitor_events().is_empty()); + assert!(nodes[0].node.get_and_clear_pending_events().is_empty()); + + // If we persist the ChannelManager here, we should get the PaymentSent event after + // deserialization. + let mut chan_manager_serialized = test_utils::TestVecWriter(Vec::new()); + if !persist_manager_post_event { + nodes[0].node.write(&mut chan_manager_serialized).unwrap(); + } + + // Now persist the ChannelMonitor and inform the ChainMonitor that we're done, generating the + // payment sent event. + chanmon_cfgs[0].persister.set_update_ret(Ok(())); + let mut chan_0_monitor_serialized = test_utils::TestVecWriter(Vec::new()); + get_monitor!(nodes[0], chan_id).write(&mut chan_0_monitor_serialized).unwrap(); + nodes[0].chain_monitor.chain_monitor.channel_monitor_updated(funding_txo, mon_updates[0]).unwrap(); + if payment_timeout { + expect_payment_failed!(nodes[0], payment_hash, true); + } else { + expect_payment_sent!(nodes[0], payment_preimage); + } + + // If we persist the ChannelManager after we get the PaymentSent event, we shouldn't get it + // twice. + if persist_manager_post_event { + nodes[0].node.write(&mut chan_manager_serialized).unwrap(); + } + + // Now reload nodes[0]... + persister = test_utils::TestPersister::new(); + let keys_manager = &chanmon_cfgs[0].keys_manager; + new_chain_monitor = test_utils::TestChainMonitor::new(Some(nodes[0].chain_source), nodes[0].tx_broadcaster.clone(), nodes[0].logger, node_cfgs[0].fee_estimator, &persister, keys_manager); + nodes[0].chain_monitor = &new_chain_monitor; + let mut chan_0_monitor_read = &chan_0_monitor_serialized.0[..]; + let (_, mut chan_0_monitor) = <(BlockHash, ChannelMonitor)>::read( + &mut chan_0_monitor_read, keys_manager).unwrap(); + assert!(chan_0_monitor_read.is_empty()); + + let (_, nodes_0_deserialized_tmp) = { + let mut channel_monitors = HashMap::new(); + channel_monitors.insert(chan_0_monitor.get_funding_txo().0, &mut chan_0_monitor); + <(BlockHash, ChannelManager)> + ::read(&mut io::Cursor::new(&chan_manager_serialized.0[..]), ChannelManagerReadArgs { + default_config: Default::default(), + keys_manager, + fee_estimator: node_cfgs[0].fee_estimator, + chain_monitor: nodes[0].chain_monitor, + tx_broadcaster: nodes[0].tx_broadcaster.clone(), + logger: nodes[0].logger, + channel_monitors, + }).unwrap() + }; + nodes_0_deserialized = nodes_0_deserialized_tmp; + + assert!(nodes[0].chain_monitor.watch_channel(chan_0_monitor.get_funding_txo().0, chan_0_monitor).is_ok()); + check_added_monitors!(nodes[0], 1); + nodes[0].node = &nodes_0_deserialized; + + if persist_manager_post_event { + assert!(nodes[0].node.get_and_clear_pending_events().is_empty()); + } else if payment_timeout { + expect_payment_failed!(nodes[0], payment_hash, true); + } else { + expect_payment_sent!(nodes[0], payment_preimage); + } + + // Note that if we re-connect the block which exposed nodes[0] to the payment preimage (but + // which the current ChannelMonitor has not seen), the ChannelManager's de-duplication of + // payment events should kick in, leaving us with no pending events here. + let height = nodes[0].blocks.lock().unwrap().len() as u32 - 1; + nodes[0].chain_monitor.chain_monitor.block_connected(&claim_block, height); + assert!(nodes[0].node.get_and_clear_pending_events().is_empty()); +} + +#[test] +fn test_dup_htlc_onchain_fails_on_reload() { + do_test_dup_htlc_onchain_fails_on_reload(true, true, true); + do_test_dup_htlc_onchain_fails_on_reload(true, true, false); + do_test_dup_htlc_onchain_fails_on_reload(true, false, false); + do_test_dup_htlc_onchain_fails_on_reload(false, true, true); + do_test_dup_htlc_onchain_fails_on_reload(false, true, false); + do_test_dup_htlc_onchain_fails_on_reload(false, false, false); +} + +#[test] +fn test_fulfill_restart_failure() { + // When we receive an update_fulfill_htlc message, we immediately consider the HTLC fully + // fulfilled. At this point, the peer can reconnect and decide to either fulfill the HTLC + // again, or fail it, giving us free money. + // + // Of course probably they won't fail it and give us free money, but because we have code to + // handle it, we should test the logic for it anyway. We do that here. + let chanmon_cfgs = create_chanmon_cfgs(2); + let node_cfgs = create_node_cfgs(2, &chanmon_cfgs); + let node_chanmgrs = create_node_chanmgrs(2, &node_cfgs, &[None, None]); + let persister: test_utils::TestPersister; + let new_chain_monitor: test_utils::TestChainMonitor; + let nodes_1_deserialized: ChannelManager; + let mut nodes = create_network(2, &node_cfgs, &node_chanmgrs); + + let chan_id = create_announced_chan_between_nodes(&nodes, 0, 1, InitFeatures::known(), InitFeatures::known()).2; + let (payment_preimage, payment_hash, _) = route_payment(&nodes[0], &[&nodes[1]], 100_000); + + // The simplest way to get a failure after a fulfill is to reload nodes[1] from a state + // pre-fulfill, which we do by serializing it here. + let mut chan_manager_serialized = test_utils::TestVecWriter(Vec::new()); + nodes[1].node.write(&mut chan_manager_serialized).unwrap(); + let mut chan_0_monitor_serialized = test_utils::TestVecWriter(Vec::new()); + get_monitor!(nodes[1], chan_id).write(&mut chan_0_monitor_serialized).unwrap(); + + nodes[1].node.claim_funds(payment_preimage); + check_added_monitors!(nodes[1], 1); + let htlc_fulfill_updates = get_htlc_update_msgs!(nodes[1], nodes[0].node.get_our_node_id()); + nodes[0].node.handle_update_fulfill_htlc(&nodes[1].node.get_our_node_id(), &htlc_fulfill_updates.update_fulfill_htlcs[0]); + expect_payment_sent!(nodes[0], payment_preimage); + + // Now reload nodes[1]... + persister = test_utils::TestPersister::new(); + let keys_manager = &chanmon_cfgs[1].keys_manager; + new_chain_monitor = test_utils::TestChainMonitor::new(Some(nodes[1].chain_source), nodes[1].tx_broadcaster.clone(), nodes[1].logger, node_cfgs[1].fee_estimator, &persister, keys_manager); + nodes[1].chain_monitor = &new_chain_monitor; + let mut chan_0_monitor_read = &chan_0_monitor_serialized.0[..]; + let (_, mut chan_0_monitor) = <(BlockHash, ChannelMonitor)>::read( + &mut chan_0_monitor_read, keys_manager).unwrap(); + assert!(chan_0_monitor_read.is_empty()); + + let (_, nodes_1_deserialized_tmp) = { + let mut channel_monitors = HashMap::new(); + channel_monitors.insert(chan_0_monitor.get_funding_txo().0, &mut chan_0_monitor); + <(BlockHash, ChannelManager)> + ::read(&mut io::Cursor::new(&chan_manager_serialized.0[..]), ChannelManagerReadArgs { + default_config: Default::default(), + keys_manager, + fee_estimator: node_cfgs[1].fee_estimator, + chain_monitor: nodes[1].chain_monitor, + tx_broadcaster: nodes[1].tx_broadcaster.clone(), + logger: nodes[1].logger, + channel_monitors, + }).unwrap() + }; + nodes_1_deserialized = nodes_1_deserialized_tmp; + + assert!(nodes[1].chain_monitor.watch_channel(chan_0_monitor.get_funding_txo().0, chan_0_monitor).is_ok()); + check_added_monitors!(nodes[1], 1); + nodes[1].node = &nodes_1_deserialized; + + nodes[0].node.peer_disconnected(&nodes[1].node.get_our_node_id(), false); + reconnect_nodes(&nodes[0], &nodes[1], (false, false), (0, 0), (0, 0), (0, 0), (0, 0), (0, 0), (false, false)); + + nodes[1].node.fail_htlc_backwards(&payment_hash); + expect_pending_htlcs_forwardable!(nodes[1]); + check_added_monitors!(nodes[1], 1); + let htlc_fail_updates = get_htlc_update_msgs!(nodes[1], nodes[0].node.get_our_node_id()); + nodes[0].node.handle_update_fail_htlc(&nodes[1].node.get_our_node_id(), &htlc_fail_updates.update_fail_htlcs[0]); + commitment_signed_dance!(nodes[0], nodes[1], htlc_fail_updates.commitment_signed, false); + // nodes[0] shouldn't generate any events here, while it just got a payment failure completion + // it had already considered the payment fulfilled, and now they just got free money. +} diff --git a/lightning/src/util/ser.rs b/lightning/src/util/ser.rs index 47bdf04f875..ba8a9fd0d35 100644 --- a/lightning/src/util/ser.rs +++ b/lightning/src/util/ser.rs @@ -502,14 +502,20 @@ impl Writeable for HashMap impl Readable for HashMap where K: Readable + Eq + Hash, - V: Readable + V: MaybeReadable { #[inline] fn read(r: &mut R) -> Result { let len: u16 = Readable::read(r)?; let mut ret = HashMap::with_capacity(len as usize); for _ in 0..len { - ret.insert(K::read(r)?, V::read(r)?); + let k = K::read(r)?; + let v_opt = V::read(r)?; + if let Some(v) = v_opt { + if ret.insert(k, v).is_some() { + return Err(DecodeError::InvalidValue); + } + } } Ok(ret) }