Prune inbound HTLC onions once forwarded

diff --git a/lightning/src/ln/channel.rs b/lightning/src/ln/channel.rs
index 8a4ef19..8a2bc30 100644
--- a/lightning/src/ln/channel.rs
+++ b/lightning/src/ln/channel.rs
@@ -318,6 +318,18 @@ impl InboundHTLCState {
 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
+	/// its onion to be pruned and no longer persisted.
+	Forwarded {
+		/// Useful if we need to fail or claim this HTLC backwards after restart, if it's missing in the
+		/// outbound edge.
+		hop_data: HTLCPreviousHopData,
+		/// Useful if we need to claim this HTLC backwards after a restart and it's missing in the
+		/// outbound edge, to generate an accurate [`Event::PaymentForwarded`].
+		///
+		/// [`Event::PaymentForwarded`]: crate::events::Event::PaymentForwarded
+		outbound_amt_msat: u64,
+	},
 	/// This HTLC was received pre-LDK 0.3, before we started persisting the onion for inbound
 	/// committed HTLCs.
 	Legacy,
@@ -328,6 +340,10 @@ impl_writeable_tlv_based_enum_upgradable!(InboundUpdateAdd,
 		(0, update_add_htlc, required),
 	},
 	(2, Legacy) => {},
+	(4, Forwarded) => {
+		(0, hop_data, required),
+		(2, outbound_amt_msat, required),
+	},
 );
 
 impl_writeable_for_vec!(&InboundUpdateAdd);
@@ -1177,6 +1193,10 @@ pub(super) struct MonitorRestoreUpdates {
 	pub channel_ready_order: ChannelReadyOrder,
 	pub announcement_sigs: Option<msgs::AnnouncementSignatures>,
 	pub tx_signatures: Option<msgs::TxSignatures>,
+	/// The sources of outbound HTLCs that were forwarded and irrevocably committed on this channel
+	/// (the outbound edge), along with their outbound amounts. Useful to store in the inbound HTLC
+	/// to ensure it gets resolved.
+	pub committed_outbound_htlc_sources: Vec<(HTLCPreviousHopData, u64)>,
 }
 
 /// The return value of `signer_maybe_unblocked`
@@ -7931,6 +7951,22 @@ where
 			.count()
 	}
 
+	/// This inbound HTLC was irrevocably forwarded to the outbound edge, so we no longer need to
+	/// persist its onion.
+	pub(super) fn prune_inbound_htlc_onion(
+		&mut self, htlc_id: u64, hop_data: HTLCPreviousHopData, outbound_amt_msat: u64,
+	) {
+		for htlc in self.context.pending_inbound_htlcs.iter_mut() {
+			if htlc.htlc_id == htlc_id {
+				if let InboundHTLCState::Committed { ref mut update_add_htlc } = htlc.state {
+					*update_add_htlc = InboundUpdateAdd::Forwarded { hop_data, outbound_amt_msat };
+					return;
+				}
+			}
+		}
+		debug_assert!(false, "If we go to prune an inbound HTLC it should be present")
+	}
+
 	/// Marks an outbound HTLC which we have received update_fail/fulfill/malformed
 	#[inline]
 	fn mark_outbound_htlc_removed(
@@ -9532,6 +9568,14 @@ where
 		mem::swap(&mut finalized_claimed_htlcs, &mut self.context.monitor_pending_finalized_fulfills);
 		let mut pending_update_adds = Vec::new();
 		mem::swap(&mut pending_update_adds, &mut self.context.monitor_pending_update_adds);
+		let committed_outbound_htlc_sources = self.context.pending_outbound_htlcs.iter().filter_map(|htlc| {
+			if let &OutboundHTLCState::LocalAnnounced(_) = &htlc.state {
+				if let HTLCSource::PreviousHopData(prev_hop_data) = &htlc.source {
+					return Some((prev_hop_data.clone(), htlc.amount_msat))
+				}
+			}
+			None
+		}).collect();
 
 		if self.context.channel_state.is_peer_disconnected() {
 			self.context.monitor_pending_revoke_and_ack = false;
@@ -9540,7 +9584,7 @@ where
 				raa: None, commitment_update: None, commitment_order: RAACommitmentOrder::RevokeAndACKFirst,
 				accepted_htlcs, failed_htlcs, finalized_claimed_htlcs, pending_update_adds,
 				funding_broadcastable, channel_ready, announcement_sigs, tx_signatures: None,
-				channel_ready_order,
+				channel_ready_order, committed_outbound_htlc_sources
 			};
 		}
 
@@ -9571,7 +9615,7 @@ where
 		MonitorRestoreUpdates {
 			raa, commitment_update, commitment_order, accepted_htlcs, failed_htlcs, finalized_claimed_htlcs,
 			pending_update_adds, funding_broadcastable, channel_ready, announcement_sigs, tx_signatures,
-			channel_ready_order,
+			channel_ready_order, committed_outbound_htlc_sources
 		}
 	}
 
diff --git a/lightning/src/ln/channelmanager.rs b/lightning/src/ln/channelmanager.rs
index 2665bf1..e50a9b8 100644
--- a/lightning/src/ln/channelmanager.rs
+++ b/lightning/src/ln/channelmanager.rs
@@ -1408,6 +1408,7 @@ enum PostMonitorUpdateChanResume {
 		decode_update_add_htlcs: Option<(u64, Vec<msgs::UpdateAddHTLC>)>,
 		finalized_claimed_htlcs: Vec<(HTLCSource, Option<AttributionData>)>,
 		failed_htlcs: Vec<(HTLCSource, PaymentHash, HTLCFailReason)>,
+		committed_outbound_htlc_sources: Vec<(HTLCPreviousHopData, u64)>,
 	},
 }
 
@@ -9586,6 +9587,7 @@ This indicates a bug inside LDK. Please report this error at https://github.com/
 		decode_update_add_htlcs: Option<(u64, Vec<msgs::UpdateAddHTLC>)>,
 		finalized_claimed_htlcs: Vec<(HTLCSource, Option<AttributionData>)>,
 		failed_htlcs: Vec<(HTLCSource, PaymentHash, HTLCFailReason)>,
+		committed_outbound_htlc_sources: Vec<(HTLCPreviousHopData, u64)>,
 	) {
 		// If the channel belongs to a batch funding transaction, the progress of the batch
 		// should be updated as we have received funding_signed and persisted the monitor.
@@ -9656,6 +9658,7 @@ This indicates a bug inside LDK. Please report this error at https://github.com/
 			};
 			self.fail_htlc_backwards_internal(&failure.0, &failure.1, &failure.2, receiver, None);
 		}
+		self.prune_persisted_inbound_htlc_onions(committed_outbound_htlc_sources);
 	}
 
 	fn handle_monitor_update_completion_actions<
@@ -10130,6 +10133,33 @@ This indicates a bug inside LDK. Please report this error at https://github.com/
 				decode_update_add_htlcs,
 				finalized_claimed_htlcs: updates.finalized_claimed_htlcs,
 				failed_htlcs: updates.failed_htlcs,
+				committed_outbound_htlc_sources: updates.committed_outbound_htlc_sources,
+			}
+		}
+	}
+
+	/// We store inbound committed HTLCs' onions in `Channel`s for use in reconstructing the pending
+	/// HTLC set on `ChannelManager` read. If an HTLC has been irrevocably forwarded to the outbound
+	/// edge, we no longer need to persist the inbound edge's onion and can prune it here.
+	fn prune_persisted_inbound_htlc_onions(
+		&self, committed_outbound_htlc_sources: Vec<(HTLCPreviousHopData, u64)>,
+	) {
+		let per_peer_state = self.per_peer_state.read().unwrap();
+		for (source, outbound_amt_msat) in committed_outbound_htlc_sources {
+			let counterparty_node_id = match source.counterparty_node_id.as_ref() {
+				Some(id) => id,
+				None => continue,
+			};
+			let mut peer_state =
+				match per_peer_state.get(counterparty_node_id).map(|state| state.lock().unwrap()) {
+					Some(peer_state) => peer_state,
+					None => continue,
+				};
+
+			if let Some(chan) =
+				peer_state.channel_by_id.get_mut(&source.channel_id).and_then(|c| c.as_funded_mut())
+			{
+				chan.prune_inbound_htlc_onion(source.htlc_id, source, outbound_amt_msat);
 			}
 		}
 	}
@@ -10144,6 +10174,18 @@ This indicates a bug inside LDK. Please report this error at https://github.com/
 		chan.test_holding_cell_outbound_htlc_forwards_count()
 	}
 
+	#[cfg(test)]
+	/// Useful to check that we prune inbound HTLC onions once they are irrevocably forwarded to the
+	/// outbound edge, see [`Self::prune_persisted_inbound_htlc_onions`].
+	pub(crate) fn test_get_inbound_committed_htlcs_with_onion(
+		&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.inbound_committed_unresolved_htlcs().len()
+	}
+
 	/// Completes channel resumption after locks have been released.
 	///
 	/// Processes the [`PostMonitorUpdateChanResume`] returned by
@@ -10169,6 +10211,7 @@ This indicates a bug inside LDK. Please report this error at https://github.com/
 				decode_update_add_htlcs,
 				finalized_claimed_htlcs,
 				failed_htlcs,
+				committed_outbound_htlc_sources,
 			} => {
 				self.post_monitor_update_unlock(
 					channel_id,
@@ -10179,6 +10222,7 @@ This indicates a bug inside LDK. Please report this error at https://github.com/
 					decode_update_add_htlcs,
 					finalized_claimed_htlcs,
 					failed_htlcs,
+					committed_outbound_htlc_sources,
 				);
 			},
 		}
diff --git a/lightning/src/ln/reload_tests.rs b/lightning/src/ln/reload_tests.rs
index 826fdbf..360ffe2 100644
--- a/lightning/src/ln/reload_tests.rs
+++ b/lightning/src/ln/reload_tests.rs
@@ -1211,6 +1211,13 @@ fn do_manager_persisted_pre_outbound_edge_forward(intercept_htlc: bool) {
 	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);
+	// While an inbound HTLC is committed in a channel but not yet forwarded, we store its onion in
+	// the `Channel` in case we need to remember it on restart. Once it's irrevocably forwarded to the
+	// outbound edge, we can prune it on the inbound edge.
+	assert_eq!(
+		nodes[1].node.test_get_inbound_committed_htlcs_with_onion(nodes[0].node.get_our_node_id(), chan_id_1),
+		1
+	);
 
 	// Decode the HTLC onion but don't forward it to the next hop, such that the HTLC ends up in
 	// `ChannelManager::forward_htlcs` or `ChannelManager::pending_intercepted_htlcs`.
@@ -1232,6 +1239,13 @@ fn do_manager_persisted_pre_outbound_edge_forward(intercept_htlc: bool) {
 	args_b_c.send_announcement_sigs = (true, true);
 	reconnect_nodes(args_b_c);
 
+	// Before an inbound HTLC is irrevocably forwarded, its onion should still be persisted within the
+	// inbound edge channel.
+	assert_eq!(
+		nodes[1].node.test_get_inbound_committed_htlcs_with_onion(nodes[0].node.get_our_node_id(), chan_id_1),
+		1
+	);
+
 	// Forward the HTLC and ensure we can claim it post-reload.
 	nodes[1].node.process_pending_htlc_forwards();
 
@@ -1254,6 +1268,12 @@ fn do_manager_persisted_pre_outbound_edge_forward(intercept_htlc: bool) {
 	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);
 	expect_and_process_pending_htlcs(&nodes[2], false);
+	// After an inbound HTLC is irrevocably forwarded, its onion should be pruned within the inbound
+	// edge channel.
+	assert_eq!(
+		nodes[1].node.test_get_inbound_committed_htlcs_with_onion(nodes[0].node.get_our_node_id(), chan_id_1),
+		0
+	);
 
 	expect_payment_claimable!(nodes[2], payment_hash, payment_secret, amt_msat, None, nodes[2].node.get_our_node_id());
 	let path: &[&[_]] = &[&[&nodes[1], &nodes[2]]];