Check pruned HTLCs were resolved on startup

diff --git a/lightning/src/ln/channel.rs b/lightning/src/ln/channel.rs
index 8a2bc30..e783f48 100644
--- a/lightning/src/ln/channel.rs
+++ b/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
@@ -7885,7 +7885,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 {
@@ -7903,13 +7905,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,
 			})
@@ -7919,18 +7919,24 @@ where
 	/// Useful when reconstructing the set of pending HTLC forwards when deserializing the
 	/// `ChannelManager`. We don't want to cache an HTLC as needing to be forwarded if it's already
 	/// present in the outbound edge, or else we'll double-forward.
-	pub(super) fn outbound_htlc_forwards(&self) -> impl Iterator<Item = HTLCPreviousHopData> + '_ {
+	pub(super) fn outbound_htlc_forwards(
+		&self,
+	) -> impl Iterator<Item = (PaymentHash, HTLCPreviousHopData)> + '_ {
 		let holding_cell_outbounds =
 			self.context.holding_cell_htlc_updates.iter().filter_map(|htlc| match htlc {
-				HTLCUpdateAwaitingACK::AddHTLC { source, .. } => match source {
-					HTLCSource::PreviousHopData(prev_hop_data) => Some(prev_hop_data.clone()),
+				HTLCUpdateAwaitingACK::AddHTLC { source, payment_hash, .. } => match source {
+					HTLCSource::PreviousHopData(prev_hop_data) => {
+						Some((*payment_hash, prev_hop_data.clone()))
+					},
 					_ => None,
 				},
 				_ => None,
 			});
 		let committed_outbounds =
 			self.context.pending_outbound_htlcs.iter().filter_map(|htlc| match &htlc.source {
-				HTLCSource::PreviousHopData(prev_hop_data) => Some(prev_hop_data.clone()),
+				HTLCSource::PreviousHopData(prev_hop_data) => {
+					Some((htlc.payment_hash, prev_hop_data.clone()))
+				},
 				_ => None,
 			});
 		holding_cell_outbounds.chain(committed_outbounds)
@@ -7967,6 +7973,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(
diff --git a/lightning/src/ln/channelmanager.rs b/lightning/src/ln/channelmanager.rs
index 569bb37..f42d294 100644
--- a/lightning/src/ln/channelmanager.rs
+++ b/lightning/src/ln/channelmanager.rs
@@ -59,9 +59,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;
@@ -10183,7 +10183,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.
@@ -18293,7 +18306,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());
 		}
@@ -18594,6 +18607,30 @@ 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: HashMap<
+			(ChannelId, PaymentHash),
+			Vec<(HTLCPreviousHopData, u64)>,
+		> = new_hash_map();
+		let prune_forwarded_htlc = |already_forwarded_htlcs: &mut HashMap<
+			(ChannelId, PaymentHash),
+			Vec<(HTLCPreviousHopData, u64)>,
+		>,
+		                            prev_hop: &HTLCPreviousHopData,
+		                            payment_hash: &PaymentHash| {
+			if let hash_map::Entry::Occupied(mut entry) =
+				already_forwarded_htlcs.entry((prev_hop.channel_id, *payment_hash))
+			{
+				entry.get_mut().retain(|(htlc, _)| prev_hop.htlc_id != htlc.htlc_id);
+				if entry.get().is_empty() {
+					entry.remove();
+				}
+			}
+		};
 		{
 			// 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
@@ -18616,16 +18653,33 @@ 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`.
+											decode_update_add_htlcs
+												.entry(scid_alias)
+												.or_insert_with(Vec::new)
+												.push(update_add_htlc);
+										},
+										InboundUpdateAdd::Forwarded {
+											hop_data,
+											outbound_amt_msat,
+										} => {
+											already_forwarded_htlcs
+												.entry((hop_data.channel_id, payment_hash))
+												.or_insert_with(Vec::new)
+												.push((hop_data, outbound_amt_msat));
+										},
+										InboundUpdateAdd::Legacy => {
+											return Err(DecodeError::InvalidValue)
+										},
+									}
 								}
 							}
 						}
@@ -18672,13 +18726,19 @@ impl<
 					if reconstruct_manager_from_monitors && !is_channel_closed {
 						if let Some(chan) = peer_state.channel_by_id.get(channel_id) {
 							if let Some(funded_chan) = chan.as_funded() {
-								for prev_hop in funded_chan.outbound_htlc_forwards() {
+								for (payment_hash, prev_hop) in funded_chan.outbound_htlc_forwards()
+								{
 									dedup_decode_update_add_htlcs(
 										&mut decode_update_add_htlcs,
 										&prev_hop,
 										"HTLC already forwarded to the outbound edge",
 										&args.logger,
 									);
+									prune_forwarded_htlc(
+										&mut already_forwarded_htlcs,
+										&prev_hop,
+										&payment_hash,
+									);
 								}
 							}
 						}
@@ -18713,6 +18773,11 @@ impl<
 										"HTLC already forwarded to the outbound edge",
 										&&logger,
 									);
+									prune_forwarded_htlc(
+										&mut already_forwarded_htlcs,
+										&prev_hop_data,
+										&htlc.payment_hash,
+									);
 								}
 
 								// The ChannelMonitor is now responsible for this HTLC's
@@ -19160,7 +19225,7 @@ impl<
 		if reconstruct_manager_from_monitors {
 			// De-duplicate HTLCs that are present in both `failed_htlcs` and `decode_update_add_htlcs`.
 			// Omitting this de-duplication could lead to redundant HTLC processing and/or bugs.
-			for (src, _, _, _, _, _) in failed_htlcs.iter() {
+			for (src, payment_hash, _, _, _, _) in failed_htlcs.iter() {
 				if let HTLCSource::PreviousHopData(prev_hop_data) = src {
 					dedup_decode_update_add_htlcs(
 						&mut decode_update_add_htlcs,
@@ -19168,6 +19233,7 @@ impl<
 						"HTLC was failed backwards during manager read",
 						&args.logger,
 					);
+					prune_forwarded_htlc(&mut already_forwarded_htlcs, prev_hop_data, payment_hash);
 				}
 			}
 
@@ -19313,9 +19379,46 @@ 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`.
+				if let Some(forwarded_htlcs) =
+					already_forwarded_htlcs.remove(&(*channel_id, payment_hash))
+				{
+					for (hop_data, outbound_amt_msat) in forwarded_htlcs {
+						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.channel_id == hop_data.channel_id)
+							});
+						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) {
@@ -19557,6 +19660,21 @@ impl<
 			channel_manager
 				.fail_htlc_backwards_internal(&source, &hash, &reason, receiver, ev_action);
 		}
+		for ((_, hash), htlcs) in already_forwarded_htlcs.into_iter() {
+			for (htlc, _) in 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,
diff --git a/lightning/src/ln/functional_test_utils.rs b/lightning/src/ln/functional_test_utils.rs
index 07f11ed..07f7c0b 100644
--- a/lightning/src/ln/functional_test_utils.rs
+++ b/lightning/src/ln/functional_test_utils.rs
@@ -1425,6 +1425,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>(
diff --git a/lightning/src/ln/reload_tests.rs b/lightning/src/ln/reload_tests.rs
index cac1871..fa0c77b 100644
--- a/lightning/src/ln/reload_tests.rs
+++ b/lightning/src/ln/reload_tests.rs
@@ -1861,3 +1861,225 @@ fn outbound_removed_holding_cell_resolved_no_double_forward() {
 	// nodes[0] should now have received the fulfill and generate PaymentSent.
 	expect_payment_sent(&nodes[0], payment_preimage, None, true, true);
 }
+
+#[test]
+fn test_reload_node_with_preimage_in_monitor_claims_htlc() {
+	// Test that if a forwarding node has an HTLC that was irrevocably removed on the outbound edge
+	// via claim but is still forwarded-and-unresolved in the inbound edge, that HTLC will not be
+	// failed back on the inbound edge on reload.
+	//
+	// For context, the ChannelManager is moving towards reconstructing the pending inbound HTLC set
+	// from Channel data on startup. If we find an inbound HTLC that is flagged as already-forwarded,
+	// we then check that the HTLC is either (a) still present in the outbound edge or (b) removed
+	// from the outbound edge but with a preimage present in the corresponding ChannelMonitor,
+	// indicating that it was removed from the outbound edge via claim. If neither of those are the
+	// case, we infer that the HTLC was removed from the outbound edge via failure and fail the HTLC
+	// backwards.
+	//
+	// Here we ensure that inbound HTLCs in case (b) above will not be failed backwards on manager
+	// reload.
+
+	let chanmon_cfgs = create_chanmon_cfgs(3);
+	let node_cfgs = create_node_cfgs(3, &chanmon_cfgs);
+	let persister;
+	let new_chain_monitor;
+	let node_chanmgrs = create_node_chanmgrs(3, &node_cfgs, &[None, None, None]);
+	let nodes_1_deserialized;
+	let mut nodes = create_network(3, &node_cfgs, &node_chanmgrs);
+
+	let node_0_id = nodes[0].node.get_our_node_id();
+	let node_1_id = nodes[1].node.get_our_node_id();
+	let node_2_id = nodes[2].node.get_our_node_id();
+
+	let chan_0_1 = create_announced_chan_between_nodes(&nodes, 0, 1);
+	let chan_1_2 = create_announced_chan_between_nodes(&nodes, 1, 2);
+
+	let chan_id_0_1 = chan_0_1.2;
+	let chan_id_1_2 = chan_1_2.2;
+
+	// Send a payment from nodes[0] to nodes[2] via nodes[1].
+	let (route, payment_hash, payment_preimage, payment_secret) =
+		get_route_and_payment_hash!(nodes[0], nodes[2], 1_000_000);
+	send_along_route_with_secret(
+		&nodes[0], route, &[&[&nodes[1], &nodes[2]]], 1_000_000, payment_hash, payment_secret,
+	);
+
+	// Claim the payment on nodes[2].
+	nodes[2].node.claim_funds(payment_preimage);
+	check_added_monitors(&nodes[2], 1);
+	expect_payment_claimed!(nodes[2], payment_hash, 1_000_000);
+
+	// Disconnect nodes[0] from nodes[1] BEFORE processing the fulfill.
+	// This prevents the claim from propagating back, leaving the inbound HTLC in ::Forwarded state.
+	nodes[0].node.peer_disconnected(node_1_id);
+	nodes[1].node.peer_disconnected(node_0_id);
+
+	// Process the fulfill from nodes[2] to nodes[1].
+	// This stores the preimage in nodes[1]'s monitor for chan_1_2.
+	let updates_2_1 = get_htlc_update_msgs(&nodes[2], &node_1_id);
+	nodes[1].node.handle_update_fulfill_htlc(node_2_id, updates_2_1.update_fulfill_htlcs[0].clone());
+	check_added_monitors(&nodes[1], 1);
+	do_commitment_signed_dance(&nodes[1], &nodes[2], &updates_2_1.commitment_signed, false, false);
+	expect_payment_forwarded!(nodes[1], nodes[0], nodes[2], Some(1000), false, false);
+
+	// Clear the holding cell's claim entry on chan_0_1 before serialization.
+	// This simulates a crash where the HTLC was fully removed from the outbound edge but is still
+	// present on the inbound edge without a resolution.
+	nodes[1].node.test_clear_channel_holding_cell(node_0_id, chan_id_0_1);
+
+	// At this point:
+	// - The inbound HTLC on nodes[1] (from nodes[0]) is in ::Forwarded state
+	// - The preimage IS in nodes[1]'s monitor for chan_1_2
+	// - The outbound HTLC to nodes[2] is resolved
+	//
+	// Serialize nodes[1] state and monitors before reloading.
+	let node_1_serialized = nodes[1].node.encode();
+	let mon_0_1_serialized = get_monitor!(nodes[1], chan_id_0_1).encode();
+	let mon_1_2_serialized = get_monitor!(nodes[1], chan_id_1_2).encode();
+
+	// Reload nodes[1].
+	// During deserialization, we track inbound HTLCs that purport to already be forwarded on the
+	// outbound edge. If any are entirely missing from the outbound edge with no preimage available,
+	// they will be failed backwards. Otherwise, as in this case where a preimage is available, the
+	// payment should be claimed backwards.
+	reload_node!(
+		nodes[1],
+		node_1_serialized,
+		&[&mon_0_1_serialized, &mon_1_2_serialized],
+		persister,
+		new_chain_monitor,
+		nodes_1_deserialized,
+		Some(true)
+	);
+
+	// When the claim is reconstructed during reload, a PaymentForwarded event is generated.
+	// This event has next_user_channel_id as None since the outbound HTLC was already removed.
+	// Fetching events triggers the pending monitor update (adding preimage) to be applied.
+	let events = nodes[1].node.get_and_clear_pending_events();
+	assert_eq!(events.len(), 1);
+	match &events[0] {
+		Event::PaymentForwarded { total_fee_earned_msat: Some(1000), .. } => {},
+		_ => panic!("Expected PaymentForwarded event"),
+	}
+	check_added_monitors(&nodes[1], 1);
+
+	// Reconnect nodes[1] to nodes[0]. The claim should be in nodes[1]'s holding cell.
+	let mut reconnect_args = ReconnectArgs::new(&nodes[1], &nodes[0]);
+	reconnect_args.pending_cell_htlc_claims = (0, 1);
+	reconnect_nodes(reconnect_args);
+
+	// nodes[0] should now have received the fulfill and generate PaymentSent.
+	expect_payment_sent(&nodes[0], payment_preimage, None, true, true);
+}
+
+#[test]
+fn test_reload_node_without_preimage_fails_htlc() {
+	// Test that if a forwarding node has an HTLC that was removed on the outbound edge via failure
+	// but is still forwarded-and-unresolved in the inbound edge, that HTLC will be correctly
+	// failed back on reload via the already_forwarded_htlcs mechanism.
+	//
+	// For context, the ChannelManager reconstructs the pending inbound HTLC set from Channel data
+	// on startup. If an inbound HTLC is present but flagged as already-forwarded, we check that
+	// the HTLC is either (a) still present in the outbound edge or (b) removed from the outbound
+	// edge but with a preimage present in the corresponding ChannelMonitor, indicating it was
+	// removed via claim. If neither, we infer the HTLC was removed via failure and fail it back.
+	//
+	// Here we test the failure case: no preimage is present, so the HTLC should be failed back.
+	let chanmon_cfgs = create_chanmon_cfgs(3);
+	let node_cfgs = create_node_cfgs(3, &chanmon_cfgs);
+	let persister;
+	let new_chain_monitor;
+	let node_chanmgrs = create_node_chanmgrs(3, &node_cfgs, &[None, None, None]);
+	let nodes_1_deserialized;
+	let mut nodes = create_network(3, &node_cfgs, &node_chanmgrs);
+
+	let node_0_id = nodes[0].node.get_our_node_id();
+	let node_1_id = nodes[1].node.get_our_node_id();
+	let node_2_id = nodes[2].node.get_our_node_id();
+
+	let chan_0_1 = create_announced_chan_between_nodes(&nodes, 0, 1);
+	let chan_1_2 = create_announced_chan_between_nodes(&nodes, 1, 2);
+
+	let chan_id_0_1 = chan_0_1.2;
+	let chan_id_1_2 = chan_1_2.2;
+
+	// Send a payment from nodes[0] to nodes[2] via nodes[1].
+	let (route, payment_hash, _, payment_secret) =
+		get_route_and_payment_hash!(nodes[0], nodes[2], 1_000_000);
+	send_along_route_with_secret(
+		&nodes[0], route, &[&[&nodes[1], &nodes[2]]], 1_000_000, payment_hash, payment_secret,
+	);
+
+	// Disconnect nodes[0] from nodes[1] BEFORE processing the failure.
+	// This prevents the fail from propagating back, leaving the inbound HTLC in ::Forwarded state.
+	nodes[0].node.peer_disconnected(node_1_id);
+	nodes[1].node.peer_disconnected(node_0_id);
+
+	// Fail the payment on nodes[2] and process the failure to nodes[1].
+	// This removes the outbound HTLC and queues a fail in the holding cell.
+	nodes[2].node.fail_htlc_backwards(&payment_hash);
+	expect_and_process_pending_htlcs_and_htlc_handling_failed(
+		&nodes[2], &[HTLCHandlingFailureType::Receive { payment_hash }]
+	);
+	check_added_monitors(&nodes[2], 1);
+
+	let updates_2_1 = get_htlc_update_msgs(&nodes[2], &node_1_id);
+	nodes[1].node.handle_update_fail_htlc(node_2_id, &updates_2_1.update_fail_htlcs[0]);
+	do_commitment_signed_dance(&nodes[1], &nodes[2], &updates_2_1.commitment_signed, false, false);
+	expect_and_process_pending_htlcs_and_htlc_handling_failed(
+		&nodes[1], &[HTLCHandlingFailureType::Forward { node_id: Some(node_2_id), channel_id: chan_id_1_2 }]
+	);
+
+	// Clear the holding cell's fail entry on chan_0_1 before serialization.
+	// This simulates a crash where the HTLC was fully removed from the outbound edge but is still
+	// present on the inbound edge without a resolution. Otherwise, we would not be able to exercise
+	// the desired failure paths due to the holding cell failure resolution being present.
+	nodes[1].node.test_clear_channel_holding_cell(node_0_id, chan_id_0_1);
+
+	// Now serialize. The state has:
+	// - Inbound HTLC on chan_0_1 in ::Forwarded state
+	// - Outbound HTLC on chan_1_2 resolved (not present)
+	// - No preimage in monitors (it was a failure)
+	// - No holding cell entry for the fail (we cleared it)
+	let node_1_serialized = nodes[1].node.encode();
+	let mon_0_1_serialized = get_monitor!(nodes[1], chan_id_0_1).encode();
+	let mon_1_2_serialized = get_monitor!(nodes[1], chan_id_1_2).encode();
+
+	// Reload nodes[1].
+	// The already_forwarded_htlcs mechanism should detect:
+	// - Inbound HTLC is in ::Forwarded state
+	// - Outbound HTLC is not present in outbound channel
+	// - No preimage in monitors
+	// Therefore it should fail the HTLC backwards.
+	reload_node!(
+		nodes[1],
+		node_1_serialized,
+		&[&mon_0_1_serialized, &mon_1_2_serialized],
+		persister,
+		new_chain_monitor,
+		nodes_1_deserialized,
+		Some(true)
+	);
+
+	// After reload, nodes[1] should have generated an HTLCHandlingFailed event.
+	let events = nodes[1].node.get_and_clear_pending_events();
+	assert!(!events.is_empty(), "Expected HTLCHandlingFailed event");
+	for event in events {
+		match event {
+			Event::HTLCHandlingFailed { .. } => {},
+			_ => panic!("Unexpected event {:?}", event),
+		}
+	}
+
+	// Process the failure so it goes back into chan_0_1's holding cell.
+	nodes[1].node.process_pending_htlc_forwards();
+	check_added_monitors(&nodes[1], 0); // No monitor update yet (peer disconnected)
+
+	// Reconnect nodes[1] to nodes[0]. The fail should be in nodes[1]'s holding cell.
+	let mut reconnect_args = ReconnectArgs::new(&nodes[1], &nodes[0]);
+	reconnect_args.pending_cell_htlc_fails = (0, 1);
+	reconnect_nodes(reconnect_args);
+
+	// nodes[0] should now have received the failure and generate PaymentFailed.
+	expect_payment_failed_conditions(&nodes[0], payment_hash, false, PaymentFailedConditions::new());
+}