De-dup decode_htlcs from monitor only if channel is closed

diff --git a/lightning/src/ln/channel.rs b/lightning/src/ln/channel.rs
index 8e69430..3678ccb 100644
--- a/lightning/src/ln/channel.rs
+++ b/lightning/src/ln/channel.rs
@@ -50,8 +50,8 @@ use crate::ln::channel_state::{
 	OutboundHTLCDetails, OutboundHTLCStateDetails,
 };
 use crate::ln::channelmanager::{
-	self, ChannelReadyOrder, FundingConfirmedMessage, HTLCFailureMsg, HTLCSource,
-	OpenChannelMessage, PaymentClaimDetails, PendingHTLCInfo, PendingHTLCStatus,
+	self, ChannelReadyOrder, FundingConfirmedMessage, HTLCFailureMsg, HTLCPreviousHopData,
+	HTLCSource, OpenChannelMessage, PaymentClaimDetails, PendingHTLCInfo, PendingHTLCStatus,
 	RAACommitmentOrder, SentHTLCId, BREAKDOWN_TIMEOUT, MAX_LOCAL_BREAKDOWN_TIMEOUT,
 	MIN_CLTV_EXPIRY_DELTA,
 };
@@ -7852,6 +7852,41 @@ where
 			.collect()
 	}
 
+	/// 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> + '_ {
+		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()),
+					_ => 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()),
+				_ => None,
+			});
+		holding_cell_outbounds.chain(committed_outbounds)
+	}
+
+	#[cfg(test)]
+	pub(super) fn test_holding_cell_outbound_htlc_forwards_count(&self) -> usize {
+		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()),
+					_ => None,
+				},
+				_ => None,
+			})
+			.count()
+	}
+
 	/// 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 13197ea..9e52282 100644
--- a/lightning/src/ln/channelmanager.rs
+++ b/lightning/src/ln/channelmanager.rs
@@ -10134,6 +10134,16 @@ This indicates a bug inside LDK. Please report this error at https://github.com/
 		}
 	}
 
+	#[cfg(test)]
+	pub(crate) fn test_holding_cell_outbound_htlc_forwards_count(
+		&self, cp_id: PublicKey, chan_id: ChannelId,
+	) -> usize {
+		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.test_holding_cell_outbound_htlc_forwards_count()
+	}
+
 	/// Completes channel resumption after locks have been released.
 	///
 	/// Processes the [`PostMonitorUpdateChanResume`] returned by
@@ -18600,6 +18610,20 @@ impl<
 					let mut peer_state_lock = peer_state_mtx.lock().unwrap();
 					let peer_state = &mut *peer_state_lock;
 					is_channel_closed = !peer_state.channel_by_id.contains_key(channel_id);
+					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() {
+									dedup_decode_update_add_htlcs(
+										&mut decode_update_add_htlcs,
+										&prev_hop,
+										"HTLC already forwarded to the outbound edge",
+										&args.logger,
+									);
+								}
+							}
+						}
+					}
 				}
 
 				for (htlc_source, (htlc, preimage_opt)) in monitor.get_all_current_outbound_htlcs()
@@ -18613,6 +18637,10 @@ impl<
 								info.prev_funding_outpoint == prev_hop_data.outpoint
 									&& info.prev_htlc_id == prev_hop_data.htlc_id
 							};
+							if !is_channel_closed {
+								continue;
+							}
+
 							// If `reconstruct_manager_from_monitors` is set, we always add all inbound committed
 							// HTLCs to `decode_update_add_htlcs` in the above loop, but we need to prune from
 							// those added HTLCs if they were already forwarded to the outbound edge. Otherwise,
@@ -18626,9 +18654,6 @@ impl<
 								);
 							}
 
-							if !is_channel_closed || reconstruct_manager_from_monitors {
-								continue;
-							}
 							// The ChannelMonitor is now responsible for this HTLC's
 							// failure/success and will let us know what its outcome is. If we
 							// still have an entry for this HTLC in `forward_htlcs_legacy`,
diff --git a/lightning/src/ln/functional_test_utils.rs b/lightning/src/ln/functional_test_utils.rs
index 2187791..6800078 100644
--- a/lightning/src/ln/functional_test_utils.rs
+++ b/lightning/src/ln/functional_test_utils.rs
@@ -1270,6 +1270,13 @@ pub fn check_added_monitors<CM: AChannelManager, H: NodeHolder<CM = CM>>(node: &
 	}
 }
 
+pub fn get_latest_mon_update_id<'a, 'b, 'c>(
+	node: &Node<'a, 'b, 'c>, channel_id: ChannelId,
+) -> (u64, u64) {
+	let monitor_id_state = node.chain_monitor.latest_monitor_update_id.lock().unwrap();
+	monitor_id_state.get(&channel_id).unwrap().clone()
+}
+
 fn claimed_htlc_matches_path<'a, 'b, 'c>(
 	origin_node: &Node<'a, 'b, 'c>, path: &[&Node<'a, 'b, 'c>], htlc: &ClaimedHTLC,
 ) -> bool {
@@ -5172,6 +5179,9 @@ pub struct ReconnectArgs<'a, 'b, 'c, 'd> {
 	pub pending_cell_htlc_claims: (usize, usize),
 	pub pending_cell_htlc_fails: (usize, usize),
 	pub pending_raa: (bool, bool),
+	/// If true, don't assert that pending messages are empty after the commitment dance completes.
+	/// Useful when holding cell HTLCs will be released and need to be handled by the caller.
+	pub allow_post_commitment_dance_msgs: (bool, bool),
 }
 
 impl<'a, 'b, 'c, 'd> ReconnectArgs<'a, 'b, 'c, 'd> {
@@ -5194,6 +5204,7 @@ impl<'a, 'b, 'c, 'd> ReconnectArgs<'a, 'b, 'c, 'd> {
 			pending_cell_htlc_claims: (0, 0),
 			pending_cell_htlc_fails: (0, 0),
 			pending_raa: (false, false),
+			allow_post_commitment_dance_msgs: (false, false),
 		}
 	}
 }
@@ -5219,6 +5230,7 @@ pub fn reconnect_nodes<'a, 'b, 'c, 'd>(args: ReconnectArgs<'a, 'b, 'c, 'd>) {
 		pending_raa,
 		pending_responding_commitment_signed,
 		pending_responding_commitment_signed_dup_monitor,
+		allow_post_commitment_dance_msgs,
 	} = args;
 	connect_nodes(node_a, node_b);
 	let reestablish_1 = get_chan_reestablish_msgs!(node_a, node_b);
@@ -5402,11 +5414,13 @@ pub fn reconnect_nodes<'a, 'b, 'c, 'd>(args: ReconnectArgs<'a, 'b, 'c, 'd>) {
 					get_event_msg!(node_a, MessageSendEvent::SendRevokeAndACK, node_b_id);
 				// No commitment_signed so get_event_msg's assert(len == 1) passes
 				node_b.node.handle_revoke_and_ack(node_a_id, &as_revoke_and_ack);
-				assert!(node_b.node.get_and_clear_pending_msg_events().is_empty());
 				check_added_monitors(
 					&node_b,
 					if pending_responding_commitment_signed_dup_monitor.0 { 0 } else { 1 },
 				);
+				if !allow_post_commitment_dance_msgs.0 {
+					assert!(node_b.node.get_and_clear_pending_msg_events().is_empty());
+				}
 			}
 		} else {
 			assert!(chan_msgs.2.is_none());
@@ -5516,11 +5530,13 @@ pub fn reconnect_nodes<'a, 'b, 'c, 'd>(args: ReconnectArgs<'a, 'b, 'c, 'd>) {
 					get_event_msg!(node_b, MessageSendEvent::SendRevokeAndACK, node_a_id);
 				// No commitment_signed so get_event_msg's assert(len == 1) passes
 				node_a.node.handle_revoke_and_ack(node_b_id, &bs_revoke_and_ack);
-				assert!(node_a.node.get_and_clear_pending_msg_events().is_empty());
 				check_added_monitors(
 					&node_a,
 					if pending_responding_commitment_signed_dup_monitor.1 { 0 } else { 1 },
 				);
+				if !allow_post_commitment_dance_msgs.1 {
+					assert!(node_a.node.get_and_clear_pending_msg_events().is_empty());
+				}
 			}
 		} else {
 			assert!(chan_msgs.2.is_none());
diff --git a/lightning/src/ln/reload_tests.rs b/lightning/src/ln/reload_tests.rs
index c043205..e6061cc 100644
--- a/lightning/src/ln/reload_tests.rs
+++ b/lightning/src/ln/reload_tests.rs
@@ -1319,6 +1319,117 @@ fn test_manager_persisted_post_outbound_edge_forward() {
 	expect_payment_sent(&nodes[0], payment_preimage, None, true, true);
 }
 
+#[test]
+fn test_manager_persisted_post_outbound_edge_holding_cell() {
+	// Test that we will not double-forward an HTLC after restart if it is already in the outbound
+	// edge's holding cell, which was previously broken.
+	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 chan_id_1 = create_announced_chan_between_nodes(&nodes, 0, 1).2;
+	let chan_id_2 = create_announced_chan_between_nodes(&nodes, 1, 2).2;
+	send_payment(&nodes[0], &[&nodes[1], &nodes[2]], 5000000);
+
+	// Lock in the HTLC from node_a <> node_b.
+	let amt_msat = 1000;
+	let (route, payment_hash, payment_preimage, payment_secret) = get_route_and_payment_hash!(nodes[0], nodes[2], amt_msat);
+	nodes[0].node.send_payment_with_route(route, payment_hash, RecipientOnionFields::secret_only(payment_secret), PaymentId(payment_hash.0)).unwrap();
+	check_added_monitors(&nodes[0], 1);
+	let updates = get_htlc_update_msgs(&nodes[0], &nodes[1].node.get_our_node_id());
+	nodes[1].node.handle_update_add_htlc(nodes[0].node.get_our_node_id(), &updates.update_add_htlcs[0]);
+	do_commitment_signed_dance(&nodes[1], &nodes[0], &updates.commitment_signed, false, false);
+
+	// Send a 2nd HTLC node_c -> node_b, to force the first HTLC into the holding cell.
+	chanmon_cfgs[1].persister.set_update_ret(ChannelMonitorUpdateStatus::InProgress);
+	let (route_2, payment_hash_2, payment_preimage_2, payment_secret_2) = get_route_and_payment_hash!(nodes[2], nodes[1], amt_msat);
+	nodes[2].node.send_payment_with_route(route_2, payment_hash_2, RecipientOnionFields::secret_only(payment_secret_2), PaymentId(payment_hash_2.0)).unwrap();
+	let send_event =
+		SendEvent::from_event(nodes[2].node.get_and_clear_pending_msg_events().remove(0));
+	nodes[1].node.handle_update_add_htlc(nodes[2].node.get_our_node_id(), &send_event.msgs[0]);
+	nodes[1].node.handle_commitment_signed_batch_test(nodes[2].node.get_our_node_id(), &send_event.commitment_msg);
+	assert!(nodes[1].node.get_and_clear_pending_msg_events().is_empty());
+	check_added_monitors(&nodes[1], 1);
+	assert!(nodes[1].node.get_and_clear_pending_msg_events().is_empty());
+
+	// Add the HTLC to the outbound edge, node_b <> node_c. Force the outbound HTLC into the b<>c
+	// holding cell.
+	nodes[1].node.process_pending_htlc_forwards();
+	check_added_monitors(&nodes[1], 0);
+	assert_eq!(
+		nodes[1].node.test_holding_cell_outbound_htlc_forwards_count(nodes[2].node.get_our_node_id(), chan_id_2),
+		1
+	);
+
+	// Disconnect peers and reload the forwarding node_b.
+	nodes[0].node.peer_disconnected(nodes[1].node.get_our_node_id());
+	nodes[2].node.peer_disconnected(nodes[1].node.get_our_node_id());
+
+	let node_b_encoded = nodes[1].node.encode();
+	let chan_0_monitor_serialized = get_monitor!(nodes[1], chan_id_1).encode();
+	let chan_1_monitor_serialized = get_monitor!(nodes[1], chan_id_2).encode();
+	reload_node!(nodes[1], node_b_encoded, &[&chan_0_monitor_serialized, &chan_1_monitor_serialized], persister, new_chain_monitor, nodes_1_deserialized);
+
+	chanmon_cfgs[1].persister.set_update_ret(ChannelMonitorUpdateStatus::Completed);
+	let (latest_update, _) = get_latest_mon_update_id(&nodes[1], chan_id_2);
+	nodes[1].chain_monitor.chain_monitor.force_channel_monitor_updated(chan_id_2, latest_update);
+
+	reconnect_nodes(ReconnectArgs::new(&nodes[1], &nodes[0]));
+
+	// Reconnect b<>c. Node_b has pending RAA + commitment_signed from the incomplete c->b
+	// commitment dance, plus an HTLC in the holding cell that will be released after the dance.
+	let mut reconnect_args = ReconnectArgs::new(&nodes[1], &nodes[2]);
+	reconnect_args.pending_raa = (false, true);
+	reconnect_args.pending_responding_commitment_signed = (false, true);
+	// Node_c needs a monitor update to catch up after processing node_b's reestablish.
+	reconnect_args.expect_renegotiated_funding_locked_monitor_update = (false, true);
+	// The holding cell HTLC will be released after the commitment dance - handle it below.
+	reconnect_args.allow_post_commitment_dance_msgs = (false, true);
+	reconnect_nodes(reconnect_args);
+
+	// The holding cell HTLC was released during the reconnect. Complete its commitment dance.
+	let holding_cell_htlc_msgs = nodes[1].node.get_and_clear_pending_msg_events();
+	assert_eq!(holding_cell_htlc_msgs.len(), 1);
+	match &holding_cell_htlc_msgs[0] {
+		MessageSendEvent::UpdateHTLCs { node_id, updates, .. } => {
+			assert_eq!(*node_id, nodes[2].node.get_our_node_id());
+			assert_eq!(updates.update_add_htlcs.len(), 1);
+			nodes[2].node.handle_update_add_htlc(nodes[1].node.get_our_node_id(), &updates.update_add_htlcs[0]);
+			do_commitment_signed_dance(&nodes[2], &nodes[1], &updates.commitment_signed, false, false);
+		}
+		_ => panic!("Unexpected message: {:?}", holding_cell_htlc_msgs[0]),
+	}
+
+	// Ensure node_b won't double-forward the outbound HTLC (this was previously broken).
+	nodes[1].node.process_pending_htlc_forwards();
+	let msgs = nodes[1].node.get_and_clear_pending_msg_events();
+	assert!(msgs.is_empty(), "Expected 0 messages, got {:?}", msgs);
+
+	// The a->b->c HTLC is now committed on node_c. The c->b HTLC is committed on node_b.
+	// Both payments should now be claimable.
+	expect_and_process_pending_htlcs(&nodes[2], false);
+	expect_payment_claimable!(nodes[2], payment_hash, payment_secret, amt_msat, None, nodes[2].node.get_our_node_id());
+	expect_payment_claimable!(nodes[1], payment_hash_2, payment_secret_2, amt_msat, None, nodes[1].node.get_our_node_id());
+
+	// Claim the a->b->c payment on node_c.
+	let path: &[&[_]] = &[&[&nodes[1], &nodes[2]]];
+	do_claim_payment_along_route(ClaimAlongRouteArgs::new(&nodes[0], path, payment_preimage));
+	expect_payment_sent(&nodes[0], payment_preimage, None, true, true);
+
+	// Claim the c->b payment on node_b.
+	nodes[1].node.claim_funds(payment_preimage_2);
+	expect_payment_claimed!(nodes[1], payment_hash_2, amt_msat);
+	check_added_monitors(&nodes[1], 1);
+	let mut update = get_htlc_update_msgs(&nodes[1], &nodes[2].node.get_our_node_id());
+	nodes[2].node.handle_update_fulfill_htlc(nodes[1].node.get_our_node_id(), update.update_fulfill_htlcs.remove(0));
+	do_commitment_signed_dance(&nodes[2], &nodes[1], &update.commitment_signed, false, false);
+	expect_payment_sent(&nodes[2], payment_preimage_2, None, true, true);
+}
+
 #[test]
 fn test_reload_partial_funding_batch() {
 	let chanmon_cfgs = create_chanmon_cfgs(3);