Track claimed outbound HTLCs in ChannelMonitors

When we receive an update_fulfill_htlc message, we immediately try
to "claim" the HTLC against the HTLCSource. If there is one, this
works great, we immediately generate a `ChannelMonitorUpdate` for
the corresponding inbound HTLC and persist that before we ever get
to processing our counterparty's `commitment_signed` and persisting
the corresponding `ChannelMonitorUpdate`.

However, if there isn't one (and this is the first successful HTLC
for a payment we sent), we immediately generate a `PaymentSent`
event and queue it up for the user. Then, a millisecond later, we
receive the `commitment_signed` from our peer, removing the HTLC
from the latest local commitment transaction as a side-effect of
the `ChannelMonitorUpdate` applied.

If the user has processed the `PaymentSent` event by that point,
great, we're done. However, if they have not, and we crash prior to
persisting the `ChannelManager`, on startup we get confused about
the state of the payment. We'll force-close the channel for being
stale, and see an HTLC which was removed and is no longer present
in the latest commitment transaction (which we're broadcasting).
Because we claim corresponding inbound HTLCs before updating a
`ChannelMonitor`, we assume such HTLCs have failed - attempting to
fail after having claimed should be a noop. However, in the
sent-payment case we now generate a `PaymentFailed` event for the
user, allowing an HTLC to complete without giving the user a
preimage.

Here we address this issue by storing the payment preimages for
claimed outbound HTLCs in the `ChannelMonitor`, in addition to the
existing inbound HTLC preimages already stored there. This allows
us to fix the specific issue described by checking for a preimage
and switching the type of event generated in response. In addition,
it reduces the risk of future confusion by ensuring we don't fail
HTLCs which were claimed but not fully committed to before a crash.

It does not, however, full fix the issue here - because the
preimages are removed after the HTLC has been fully removed from
available commitment transactions if we are substantially delayed
in persisting the `ChannelManager` from the time we receive the
`update_fulfill_htlc` until after a full commitment signed dance
completes we may still hit this issue. The full fix for this issue
is to delay the persistence of the `ChannelMonitorUpdate` until
after the `PaymentSent` event has been processed. This avoids the
issue entirely, ensuring we process the event before updating the
`ChannelMonitor`, the same as we ensure the upstream HTLC has been
claimed before updating the `ChannelMonitor` for forwarded
payments.

The full solution will be implemented in a later work, however this
change still makes sense at that point as well - if we were to
delay the initial `commitment_signed` `ChannelMonitorUpdate` util
after the `PaymentSent` event has been processed (which likely
requires a database update on the users' end), we'd hold our
`commitment_signed` + `revoke_and_ack` response for two DB writes
(i.e. `fsync()` calls), making our commitment transaction
processing a full `fsync` slower. By making this change first, we
can instead delay the `ChannelMonitorUpdate` from the
counterparty's final `revoke_and_ack` message until the event has
been processed, giving us a full network roundtrip to do so and
avoiding delaying our response as long as an `fsync` is faster than
a network roundtrip.

Author: TheBlueMatt

lightning/src/chain/channelmonitor.rs

@@ -37,7 +37,7 @@ use crate::ln::{PaymentHash, PaymentPreimage};
 use crate::ln::msgs::DecodeError;
 use crate::ln::chan_utils;
 use crate::ln::chan_utils::{CounterpartyCommitmentSecrets, HTLCOutputInCommitment, HTLCClaim, ChannelTransactionParameters, HolderCommitmentTransaction};
-use crate::ln::channelmanager::HTLCSource;
+use crate::ln::channelmanager::{HTLCSource, SentHTLCId};
 use crate::chain;
 use crate::chain::{BestBlock, WatchedOutput};
 use crate::chain::chaininterface::{BroadcasterInterface, FeeEstimator, LowerBoundedFeeEstimator};
@@ -498,6 +498,7 @@ pub(crate) enum ChannelMonitorUpdateStep {
 	LatestHolderCommitmentTXInfo {
 		commitment_tx: HolderCommitmentTransaction,
 		htlc_outputs: Vec<(HTLCOutputInCommitment, Option<Signature>, Option<HTLCSource>)>,
+		claimed_htlcs: Vec<(SentHTLCId, PaymentPreimage)>,
 	},
 	LatestCounterpartyCommitmentTXInfo {
 		commitment_txid: Txid,
@@ -540,6 +541,7 @@ impl ChannelMonitorUpdateStep {
 impl_writeable_tlv_based_enum_upgradable!(ChannelMonitorUpdateStep,
 	(0, LatestHolderCommitmentTXInfo) => {
 		(0, commitment_tx, required),
+		(1, claimed_htlcs, vec_type),
 		(2, htlc_outputs, vec_type),
 	},
 	(1, LatestCounterpartyCommitmentTXInfo) => {
@@ -754,6 +756,8 @@ pub(crate) struct ChannelMonitorImpl<Signer: WriteableEcdsaChannelSigner> {
 	/// Serialized to disk but should generally not be sent to Watchtowers.
 	counterparty_hash_commitment_number: HashMap<PaymentHash, u64>,
 
+	counterparty_fulfilled_htlcs: HashMap<SentHTLCId, PaymentPreimage>,
+
 	// We store two holder commitment transactions to avoid any race conditions where we may update
 	// some monitors (potentially on watchtowers) but then fail to update others, resulting in the
 	// various monitors for one channel being out of sync, and us broadcasting a holder
@@ -1033,6 +1037,7 @@ impl<Signer: WriteableEcdsaChannelSigner> Writeable for ChannelMonitorImpl<Signe
 			(9, self.counterparty_node_id, option),
 			(11, self.confirmed_commitment_tx_counterparty_output, option),
 			(13, self.spendable_txids_confirmed, vec_type),
+			(15, self.counterparty_fulfilled_htlcs, required),
 		});
 
 		Ok(())
@@ -1120,6 +1125,7 @@ impl<Signer: WriteableEcdsaChannelSigner> ChannelMonitor<Signer> {
 			counterparty_claimable_outpoints: HashMap::new(),
 			counterparty_commitment_txn_on_chain: HashMap::new(),
 			counterparty_hash_commitment_number: HashMap::new(),
+			counterparty_fulfilled_htlcs: HashMap::new(),
 
 			prev_holder_signed_commitment_tx: None,
 			current_holder_commitment_tx: holder_commitment_tx,
@@ -1174,7 +1180,7 @@ impl<Signer: WriteableEcdsaChannelSigner> ChannelMonitor<Signer> {
 		&self, holder_commitment_tx: HolderCommitmentTransaction,
 		htlc_outputs: Vec<(HTLCOutputInCommitment, Option<Signature>, Option<HTLCSource>)>,
 	) -> Result<(), ()> {
-		self.inner.lock().unwrap().provide_latest_holder_commitment_tx(holder_commitment_tx, htlc_outputs).map_err(|_| ())
+		self.inner.lock().unwrap().provide_latest_holder_commitment_tx(holder_commitment_tx, htlc_outputs, &Vec::new()).map_err(|_| ())
 	}
 
 	/// This is used to provide payment preimage(s) out-of-band during startup without updating the
@@ -1812,7 +1818,7 @@ impl<Signer: WriteableEcdsaChannelSigner> ChannelMonitor<Signer> {
 	///
 	/// This is similar to [`Self::get_pending_outbound_htlcs`] except it includes HTLCs which were
 	/// resolved by this `ChannelMonitor`.
-	pub(crate) fn get_all_current_outbound_htlcs(&self) -> HashMap<HTLCSource, HTLCOutputInCommitment> {
+	pub(crate) fn get_all_current_outbound_htlcs(&self) -> HashMap<HTLCSource, (HTLCOutputInCommitment, Option<PaymentPreimage>)> {
 		let mut res = HashMap::new();
 		// Just examine the available counterparty commitment transactions. See docs on
 		// `fail_unbroadcast_htlcs`, below, for justification.
@@ -1822,7 +1828,8 @@ impl<Signer: WriteableEcdsaChannelSigner> ChannelMonitor<Signer> {
 				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());
+							res.insert((**source).clone(), (htlc.clone(),
+								us.counterparty_fulfilled_htlcs.get(&SentHTLCId::from_source(source)).cloned()));
 						}
 					}
 				}
@@ -1839,7 +1846,7 @@ impl<Signer: WriteableEcdsaChannelSigner> ChannelMonitor<Signer> {
 
 	/// 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<HTLCSource, HTLCOutputInCommitment> {
+	pub(crate) fn get_pending_outbound_htlcs(&self) -> HashMap<HTLCSource, (HTLCOutputInCommitment, Option<PaymentPreimage>)> {
 		let us = self.inner.lock().unwrap();
 		// 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,
@@ -1888,7 +1895,10 @@ impl<Signer: WriteableEcdsaChannelSigner> ChannelMonitor<Signer> {
 							} else { false }
 						});
 						if !htlc_update_confd {
-							res.insert(source.clone(), htlc.clone());
+							if us.counterparty_fulfilled_htlcs.get(&SentHTLCId::from_source(source)).is_none() {
+								res.insert(source.clone(), (htlc.clone(),
+									us.counterparty_fulfilled_htlcs.get(&SentHTLCId::from_source(source)).cloned()));
+							}
 						}
 					}
 				}
@@ -1970,6 +1980,9 @@ macro_rules! fail_unbroadcast_htlcs {
 								}
 							}
 							if matched_htlc { continue; }
+							if $self.counterparty_fulfilled_htlcs.get(&SentHTLCId::from_source(source)).is_some() {
+								continue;
+							}
 							$self.onchain_events_awaiting_threshold_conf.retain(|ref entry| {
 								if entry.height != $commitment_tx_conf_height { return true; }
 								match entry.event {
@@ -2041,8 +2054,23 @@ impl<Signer: WriteableEcdsaChannelSigner> ChannelMonitorImpl<Signer> {
 		// Prune HTLCs from the previous counterparty commitment tx so we don't generate failure/fulfill
 		// events for now-revoked/fulfilled HTLCs.
 		if let Some(txid) = self.prev_counterparty_commitment_txid.take() {
-			for &mut (_, ref mut source) in self.counterparty_claimable_outpoints.get_mut(&txid).unwrap() {
-				*source = None;
+			let cur_claimables = self.counterparty_claimable_outpoints.get(
+				&self.current_counterparty_commitment_txid.unwrap()).unwrap();
+			for (_, ref source_opt) in self.counterparty_claimable_outpoints.get(&txid).unwrap() {
+				if let Some(source) = source_opt {
+					let mut also_in_cur_commitment = false;
+					for (_, ref cur_source_opt) in cur_claimables {
+						if cur_source_opt == source_opt {
+							also_in_cur_commitment = true;
+						}
+					}
+					if !also_in_cur_commitment {
+						self.counterparty_fulfilled_htlcs.remove(&SentHTLCId::from_source(source));
+					}
+				}
+			}
+			for &mut (_, ref mut source_opt) in self.counterparty_claimable_outpoints.get_mut(&txid).unwrap() {
+				*source_opt = None;
 			}
 		}
 
@@ -2127,8 +2155,8 @@ impl<Signer: WriteableEcdsaChannelSigner> ChannelMonitorImpl<Signer> {
 	/// is important that any clones of this channel monitor (including remote clones) by kept
 	/// up-to-date as our holder commitment transaction is updated.
 	/// Panics if set_on_holder_tx_csv has never been called.
-	fn provide_latest_holder_commitment_tx(&mut self, holder_commitment_tx: HolderCommitmentTransaction, htlc_outputs: Vec<(HTLCOutputInCommitment, Option<Signature>, Option<HTLCSource>)>) -> Result<(), &'static str> {
-		// block for Rust 1.34 compat
+	fn provide_latest_holder_commitment_tx(&mut self, holder_commitment_tx: HolderCommitmentTransaction, htlc_outputs: Vec<(HTLCOutputInCommitment, Option<Signature>, Option<HTLCSource>)>, claimed_htlcs: &[(SentHTLCId, PaymentPreimage)]) -> Result<(), &'static str> {
+		// block for Rust 1.34 compa
 		let mut new_holder_commitment_tx = {
 			let trusted_tx = holder_commitment_tx.trust();
 			let txid = trusted_tx.txid();
@@ -2149,6 +2177,22 @@ impl<Signer: WriteableEcdsaChannelSigner> ChannelMonitorImpl<Signer> {
 		self.onchain_tx_handler.provide_latest_holder_tx(holder_commitment_tx);
 		mem::swap(&mut new_holder_commitment_tx, &mut self.current_holder_commitment_tx);
 		self.prev_holder_signed_commitment_tx = Some(new_holder_commitment_tx);
+		for (claimed_htlc_id, claimed_preimage) in claimed_htlcs {
+			#[cfg(debug_assertions)] {
+				let mut found_matching_pending_htlc = false;
+				for (_, source_opt) in self.counterparty_claimable_outpoints.get(
+					&self.current_counterparty_commitment_txid.unwrap()
+				).unwrap().iter() {
+					if let Some(source) = source_opt {
+						if SentHTLCId::from_source(source) == *claimed_htlc_id {
+							found_matching_pending_htlc = true;
+						}
+					}
+				}
+				assert!(found_matching_pending_htlc);
+			}
+			self.counterparty_fulfilled_htlcs.insert(*claimed_htlc_id, *claimed_preimage);
+		}
 		if self.holder_tx_signed {
 			return Err("Latest holder commitment signed has already been signed, update is rejected");
 		}
@@ -2243,10 +2287,10 @@ impl<Signer: WriteableEcdsaChannelSigner> ChannelMonitorImpl<Signer> {
 		let bounded_fee_estimator = LowerBoundedFeeEstimator::new(&*fee_estimator);
 		for update in updates.updates.iter() {
 			match update {
-				ChannelMonitorUpdateStep::LatestHolderCommitmentTXInfo { commitment_tx, htlc_outputs } => {
+				ChannelMonitorUpdateStep::LatestHolderCommitmentTXInfo { commitment_tx, htlc_outputs, claimed_htlcs } => {
 					log_trace!(logger, "Updating ChannelMonitor with latest holder commitment transaction info");
 					if self.lockdown_from_offchain { panic!(); }
-					if let Err(e) = self.provide_latest_holder_commitment_tx(commitment_tx.clone(), htlc_outputs.clone()) {
+					if let Err(e) = self.provide_latest_holder_commitment_tx(commitment_tx.clone(), htlc_outputs.clone(), &claimed_htlcs) {
 						log_error!(logger, "Providing latest holder commitment transaction failed/was refused:");
 						log_error!(logger, "    {}", e);
 						ret = Err(());
@@ -3868,6 +3912,7 @@ impl<'a, 'b, ES: EntropySource, SP: SignerProvider> ReadableArgs<(&'a ES, &'b SP
 		let mut counterparty_node_id = None;
 		let mut confirmed_commitment_tx_counterparty_output = None;
 		let mut spendable_txids_confirmed = Some(Vec::new());
+		let mut counterparty_fulfilled_htlcs = Some(HashMap::new());
 		read_tlv_fields!(reader, {
 			(1, funding_spend_confirmed, option),
 			(3, htlcs_resolved_on_chain, vec_type),
@@ -3876,6 +3921,7 @@ impl<'a, 'b, ES: EntropySource, SP: SignerProvider> ReadableArgs<(&'a ES, &'b SP
 			(9, counterparty_node_id, option),
 			(11, confirmed_commitment_tx_counterparty_output, option),
 			(13, spendable_txids_confirmed, vec_type),
+			(15, counterparty_fulfilled_htlcs, option),
 		});
 
 		Ok((best_block.block_hash(), ChannelMonitor::from_impl(ChannelMonitorImpl {
@@ -3904,6 +3950,7 @@ impl<'a, 'b, ES: EntropySource, SP: SignerProvider> ReadableArgs<(&'a ES, &'b SP
 			counterparty_claimable_outpoints,
 			counterparty_commitment_txn_on_chain,
 			counterparty_hash_commitment_number,
+			counterparty_fulfilled_htlcs: counterparty_fulfilled_htlcs.unwrap(),
 
 			prev_holder_signed_commitment_tx,
 			current_holder_commitment_tx,

lightning/src/ln/channel.rs

@@ -27,7 +27,7 @@ use crate::ln::features::{ChannelTypeFeatures, InitFeatures};
 use crate::ln::msgs;
 use crate::ln::msgs::{DecodeError, OptionalField, DataLossProtect};
 use crate::ln::script::{self, ShutdownScript};
-use crate::ln::channelmanager::{self, CounterpartyForwardingInfo, PendingHTLCStatus, HTLCSource, HTLCFailureMsg, PendingHTLCInfo, RAACommitmentOrder, BREAKDOWN_TIMEOUT, MIN_CLTV_EXPIRY_DELTA, MAX_LOCAL_BREAKDOWN_TIMEOUT};
+use crate::ln::channelmanager::{self, CounterpartyForwardingInfo, PendingHTLCStatus, HTLCSource, SentHTLCId, HTLCFailureMsg, PendingHTLCInfo, RAACommitmentOrder, BREAKDOWN_TIMEOUT, MIN_CLTV_EXPIRY_DELTA, MAX_LOCAL_BREAKDOWN_TIMEOUT};
 use crate::ln::chan_utils::{CounterpartyCommitmentSecrets, TxCreationKeys, HTLCOutputInCommitment, htlc_success_tx_weight, htlc_timeout_tx_weight, make_funding_redeemscript, ChannelPublicKeys, CommitmentTransaction, HolderCommitmentTransaction, ChannelTransactionParameters, CounterpartyChannelTransactionParameters, MAX_HTLCS, get_commitment_transaction_number_obscure_factor, ClosingTransaction};
 use crate::ln::chan_utils;
 use crate::ln::onion_utils::HTLCFailReason;
@@ -192,6 +192,7 @@ enum OutboundHTLCState {
 
 #[derive(Clone)]
 enum OutboundHTLCOutcome {
+	/// LDK version 0.0.105+ will always fill in the preimage here.
 	Success(Option<PaymentPreimage>),
 	Failure(HTLCFailReason),
 }
@@ -3159,15 +3160,6 @@ impl<Signer: WriteableEcdsaChannelSigner> Channel<Signer> {
 			}
 		}
 
-		self.latest_monitor_update_id += 1;
-		let mut monitor_update = ChannelMonitorUpdate {
-			update_id: self.latest_monitor_update_id,
-			updates: vec![ChannelMonitorUpdateStep::LatestHolderCommitmentTXInfo {
-				commitment_tx: holder_commitment_tx,
-				htlc_outputs: htlcs_and_sigs
-			}]
-		};
-
 		for htlc in self.pending_inbound_htlcs.iter_mut() {
 			let new_forward = if let &InboundHTLCState::RemoteAnnounced(ref forward_info) = &htlc.state {
 				Some(forward_info.clone())
@@ -3179,18 +3171,36 @@ impl<Signer: WriteableEcdsaChannelSigner> Channel<Signer> {
 				need_commitment = true;
 			}
 		}
+		let mut claimed_htlcs = Vec::new();
 		for htlc in self.pending_outbound_htlcs.iter_mut() {
 			if let &mut OutboundHTLCState::RemoteRemoved(ref mut outcome) = &mut htlc.state {
 				log_trace!(logger, "Updating HTLC {} to AwaitingRemoteRevokeToRemove due to commitment_signed in channel {}.",
 					log_bytes!(htlc.payment_hash.0), log_bytes!(self.channel_id));
 				// Grab the preimage, if it exists, instead of cloning
 				let mut reason = OutboundHTLCOutcome::Success(None);
 				mem::swap(outcome, &mut reason);
+				if let OutboundHTLCOutcome::Success(Some(preimage)) = reason {
+					// While failing to handle the `Success(None)` case here could result in
+					// forgetting some HTLC claims, it is only reachable if the HTLC was claimed on
+					// LDK 0.0.104 and prior versions, then upgrading to a modern LDK before the
+					// HTLC fully resolves.
+					claimed_htlcs.push((SentHTLCId::from_source(&htlc.source), preimage));
+				}
 				htlc.state = OutboundHTLCState::AwaitingRemoteRevokeToRemove(reason);
 				need_commitment = true;
 			}
 		}
 
+		self.latest_monitor_update_id += 1;
+		let mut monitor_update = ChannelMonitorUpdate {
+			update_id: self.latest_monitor_update_id,
+			updates: vec![ChannelMonitorUpdateStep::LatestHolderCommitmentTXInfo {
+				commitment_tx: holder_commitment_tx,
+				htlc_outputs: htlcs_and_sigs,
+				claimed_htlcs,
+			}]
+		};
+
 		self.cur_holder_commitment_transaction_number -= 1;
 		// Note that if we need_commitment & !AwaitingRemoteRevoke we'll call
 		// build_commitment_no_status_check() next which will reset this to RAAFirst.