Check pruned HTLCs were resolved on startup

In a recent commit, we added support for pruning an inbound HTLC's persisted
onion once the HTLC has been irrevocably forwarded to the outbound edge.

Here, we add a check on startup that those inbound HTLCs were actually handled.
Specifically, we check that the inbound HTLC is either (a) currently present in
the outbound edge or (b) was removed via claim. If neither of those are true,
we infer that the HTLC was removed from the outbound edge via fail and fail the
inbound HTLC backwards.

Author: valentinewallace

lightning/src/ln/channel.rs

@@ -314,8 +314,8 @@ impl InboundHTLCState {
 /// `ChannelManager` persist.
 ///
 /// Useful for reconstructing the pending HTLC set on startup.
-#[derive(Debug)]
-enum InboundUpdateAdd {
+#[derive(Debug, Clone)]
+pub(super) enum InboundUpdateAdd {
 	/// The inbound committed HTLC's update_add_htlc message.
 	WithOnion { update_add_htlc: msgs::UpdateAddHTLC },
 	/// This inbound HTLC is a forward that was irrevocably committed to the outbound edge, allowing
@@ -7862,7 +7862,9 @@ where
 	}
 
 	/// Useful for reconstructing the set of pending HTLCs when deserializing the `ChannelManager`.
-	pub(super) fn inbound_committed_unresolved_htlcs(&self) -> Vec<msgs::UpdateAddHTLC> {
+	pub(super) fn inbound_committed_unresolved_htlcs(
+		&self,
+	) -> Vec<(PaymentHash, InboundUpdateAdd)> {
 		// We don't want to return an HTLC as needing processing if it already has a resolution that's
 		// pending in the holding cell.
 		let htlc_resolution_in_holding_cell = |id: u64| -> bool {
@@ -7880,13 +7882,11 @@ where
 			.pending_inbound_htlcs
 			.iter()
 			.filter_map(|htlc| match &htlc.state {
-				InboundHTLCState::Committed {
-					update_add_htlc: InboundUpdateAdd::WithOnion { update_add_htlc },
-				} => {
+				InboundHTLCState::Committed { update_add_htlc } => {
 					if htlc_resolution_in_holding_cell(htlc.htlc_id) {
 						return None;
 					}
-					Some(update_add_htlc.clone())
+					Some((htlc.payment_hash, update_add_htlc.clone()))
 				},
 				_ => None,
 			})
@@ -7944,6 +7944,12 @@ where
 		debug_assert!(false, "If we go to prune an inbound HTLC it should be present")
 	}
 
+	/// Useful for testing crash scenarios where the holding cell is not persisted.
+	#[cfg(test)]
+	pub(super) fn test_clear_holding_cell(&mut self) {
+		self.context.holding_cell_htlc_updates.clear()
+	}
+
 	/// Marks an outbound HTLC which we have received update_fail/fulfill/malformed
 	#[inline]
 	fn mark_outbound_htlc_removed(

lightning/src/ln/channelmanager.rs

@@ -58,9 +58,9 @@ use crate::ln::chan_utils::selected_commitment_sat_per_1000_weight;
 use crate::ln::channel::QuiescentAction;
 use crate::ln::channel::{
 	self, hold_time_since, Channel, ChannelError, ChannelUpdateStatus, DisconnectResult,
-	FundedChannel, FundingTxSigned, InboundV1Channel, OutboundV1Channel, PendingV2Channel,
-	ReconnectionMsg, ShutdownResult, SpliceFundingFailed, StfuResponse, UpdateFulfillCommitFetch,
-	WithChannelContext,
+	FundedChannel, FundingTxSigned, InboundUpdateAdd, InboundV1Channel, OutboundV1Channel,
+	PendingV2Channel, ReconnectionMsg, ShutdownResult, SpliceFundingFailed, StfuResponse,
+	UpdateFulfillCommitFetch, WithChannelContext,
 };
 use crate::ln::channel_state::ChannelDetails;
 use crate::ln::funding::SpliceContribution;
@@ -10145,7 +10145,20 @@ This indicates a bug inside LDK. Please report this error at https://github.com/
 		let per_peer_state = self.per_peer_state.read().unwrap();
 		let peer_state = per_peer_state.get(&cp_id).map(|state| state.lock().unwrap()).unwrap();
 		let chan = peer_state.channel_by_id.get(&chan_id).and_then(|c| c.as_funded()).unwrap();
-		chan.inbound_committed_unresolved_htlcs().len()
+		chan.inbound_committed_unresolved_htlcs()
+			.iter()
+			.filter(|(_, htlc)| matches!(htlc, InboundUpdateAdd::WithOnion { .. }))
+			.count()
+	}
+
+	#[cfg(test)]
+	/// Useful for testing crash scenarios where the holding cell of a channel is not persisted.
+	pub(crate) fn test_clear_channel_holding_cell(&self, cp_id: PublicKey, chan_id: ChannelId) {
+		let per_peer_state = self.per_peer_state.read().unwrap();
+		let mut peer_state = per_peer_state.get(&cp_id).map(|state| state.lock().unwrap()).unwrap();
+		let chan =
+			peer_state.channel_by_id.get_mut(&chan_id).and_then(|c| c.as_funded_mut()).unwrap();
+		chan.test_clear_holding_cell();
 	}
 
 	/// Completes channel resumption after locks have been released.
@@ -18158,7 +18171,7 @@ impl<
 		}
 
 		// Post-deserialization processing
-		let mut decode_update_add_htlcs = new_hash_map();
+		let mut decode_update_add_htlcs: HashMap<u64, Vec<msgs::UpdateAddHTLC>> = new_hash_map();
 		if fake_scid_rand_bytes.is_none() {
 			fake_scid_rand_bytes = Some(args.entropy_source.get_secure_random_bytes());
 		}
@@ -18450,6 +18463,22 @@ impl<
 		// have a fully-constructed `ChannelManager` at the end.
 		let mut pending_claims_to_replay = Vec::new();
 
+		// If we find an inbound HTLC that claims to already be forwarded to the outbound edge, we
+		// store an identifier for it here and verify that it is either (a) present in the outbound
+		// edge or (b) removed from the outbound edge via claim. If it's in neither of these states, we
+		// infer that it was removed from the outbound edge via fail, and fail it backwards to ensure
+		// that it is handled.
+		let mut already_forwarded_htlcs = Vec::new();
+		let prune_forwarded_htlc =
+			|already_forwarded_htlcs: &mut Vec<(PaymentHash, HTLCPreviousHopData, u64)>,
+			 prev_hop: &HTLCPreviousHopData| {
+				if let Some(idx) = already_forwarded_htlcs.iter().position(|(_, htlc, _)| {
+					prev_hop.htlc_id == htlc.htlc_id
+						&& prev_hop.prev_outbound_scid_alias == htlc.prev_outbound_scid_alias
+				}) {
+					already_forwarded_htlcs.swap_remove(idx);
+				}
+			};
 		{
 			// 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
@@ -18472,16 +18501,38 @@ impl<
 					if reconstruct_manager_from_monitors {
 						if let Some(chan) = peer_state.channel_by_id.get(channel_id) {
 							if let Some(funded_chan) = chan.as_funded() {
+								let scid_alias = funded_chan.context.outbound_scid_alias();
 								let inbound_committed_update_adds =
 									funded_chan.inbound_committed_unresolved_htlcs();
-								if !inbound_committed_update_adds.is_empty() {
-									// Reconstruct `ChannelManager::decode_update_add_htlcs` from the serialized
-									// `Channel`, as part of removing the requirement to regularly persist the
-									// `ChannelManager`.
-									decode_update_add_htlcs.insert(
-										funded_chan.context.outbound_scid_alias(),
-										inbound_committed_update_adds,
-									);
+								for (payment_hash, htlc) in inbound_committed_update_adds {
+									match htlc {
+										InboundUpdateAdd::WithOnion { update_add_htlc } => {
+											// Reconstruct `ChannelManager::decode_update_add_htlcs` from the serialized
+											// `Channel` as part of removing the requirement to regularly persist the
+											// `ChannelManager`.
+											match decode_update_add_htlcs.entry(scid_alias) {
+												hash_map::Entry::Occupied(mut entry) => {
+													entry.get_mut().push(update_add_htlc);
+												},
+												hash_map::Entry::Vacant(entry) => {
+													entry.insert(vec![update_add_htlc]);
+												},
+											}
+										},
+										InboundUpdateAdd::Forwarded {
+											hop_data,
+											outbound_amt_msat,
+										} => {
+											already_forwarded_htlcs.push((
+												payment_hash,
+												hop_data,
+												outbound_amt_msat,
+											));
+										},
+										InboundUpdateAdd::Legacy => {
+											return Err(DecodeError::InvalidValue)
+										},
+									}
 								}
 							}
 						}
@@ -18535,6 +18586,7 @@ impl<
 										"HTLC already forwarded to the outbound edge",
 										&args.logger,
 									);
+									prune_forwarded_htlc(&mut already_forwarded_htlcs, &prev_hop);
 								}
 							}
 						}
@@ -18569,6 +18621,10 @@ impl<
 										"HTLC already forwarded to the outbound edge",
 										&&logger,
 									);
+									prune_forwarded_htlc(
+										&mut already_forwarded_htlcs,
+										&prev_hop_data,
+									);
 								}
 
 								// The ChannelMonitor is now responsible for this HTLC's
@@ -19093,6 +19149,7 @@ impl<
 						"HTLC was failed backwards during manager read",
 						&args.logger,
 					);
+					prune_forwarded_htlc(&mut already_forwarded_htlcs, prev_hop_data);
 				}
 			}
 
@@ -19238,9 +19295,47 @@ impl<
 		};
 
 		let mut processed_claims: HashSet<Vec<MPPClaimHTLCSource>> = new_hash_set();
-		for (_, monitor) in args.channel_monitors.iter() {
+		for (channel_id, monitor) in args.channel_monitors.iter() {
 			for (payment_hash, (payment_preimage, payment_claims)) in monitor.get_stored_preimages()
 			{
+				// If we have unresolved inbound committed HTLCs that were already forwarded to the
+				// outbound edge and removed via claim, we need to make sure to claim them backwards via
+				// adding them to `pending_claims_to_replay`.
+				for (hash, hop_data, outbound_amt_msat) in
+					mem::take(&mut already_forwarded_htlcs).drain(..)
+				{
+					if hash != payment_hash {
+						already_forwarded_htlcs.push((hash, hop_data, outbound_amt_msat));
+						continue;
+					}
+					let new_pending_claim = !pending_claims_to_replay.iter().any(|(src, _, _, _, _, _, _)| {
+						matches!(src, HTLCSource::PreviousHopData(hop) if hop.htlc_id == hop_data.htlc_id && hop.prev_outbound_scid_alias == hop_data.prev_outbound_scid_alias)
+					});
+					if new_pending_claim {
+						let counterparty_node_id = monitor.get_counterparty_node_id();
+						let is_channel_closed = channel_manager
+							.per_peer_state
+							.read()
+							.unwrap()
+							.get(&counterparty_node_id)
+							.map_or(true, |peer_state_mtx| {
+								!peer_state_mtx
+									.lock()
+									.unwrap()
+									.channel_by_id
+									.contains_key(channel_id)
+							});
+						pending_claims_to_replay.push((
+							HTLCSource::PreviousHopData(hop_data),
+							payment_preimage,
+							outbound_amt_msat,
+							is_channel_closed,
+							counterparty_node_id,
+							monitor.get_funding_txo(),
+							*channel_id,
+						));
+					}
+				}
 				if !payment_claims.is_empty() {
 					for payment_claim in payment_claims {
 						if processed_claims.contains(&payment_claim.mpp_parts) {
@@ -19482,6 +19577,18 @@ impl<
 			channel_manager
 				.fail_htlc_backwards_internal(&source, &hash, &reason, receiver, ev_action);
 		}
+		for (hash, htlc, _) in already_forwarded_htlcs {
+			let channel_id = htlc.channel_id;
+			let node_id = htlc.counterparty_node_id;
+			let source = HTLCSource::PreviousHopData(htlc);
+			let failure_reason = LocalHTLCFailureReason::TemporaryChannelFailure;
+			let failure_data = channel_manager.get_htlc_inbound_temp_fail_data(failure_reason);
+			let reason = HTLCFailReason::reason(failure_reason, failure_data);
+			let receiver = HTLCHandlingFailureType::Forward { node_id, channel_id };
+			// The event completion action is only relevant for HTLCs that originate from our node, not
+			// forwarded HTLCs.
+			channel_manager.fail_htlc_backwards_internal(&source, &hash, &reason, receiver, None);
+		}
 
 		for (
 			source,

lightning/src/ln/functional_test_utils.rs

@@ -1423,6 +1423,23 @@ macro_rules! reload_node {
 			None
 		);
 	};
+	// Reload the node and have the `ChannelManager` use new codepaths that reconstruct its set of
+	// pending HTLCs from `Channel{Monitor}` data.
+	($node: expr, $chanman_encoded: expr, $monitors_encoded: expr, $persister:
+	 ident, $new_chain_monitor: ident, $new_channelmanager: ident, $reconstruct_pending_htlcs: expr
+	) => {
+		let config = $node.node.get_current_config();
+		_reload_node_inner!(
+			$node,
+			config,
+			$chanman_encoded,
+			$monitors_encoded,
+			$persister,
+			$new_chain_monitor,
+			$new_channelmanager,
+			$reconstruct_pending_htlcs
+		);
+	};
 }
 
 pub fn create_funding_transaction<'a, 'b, 'c>(