ln/events: multiple htlcs in/out for trampoline PaymentForwarded

diff --git a/lightning-liquidity/tests/lsps2_integration_tests.rs b/lightning-liquidity/tests/lsps2_integration_tests.rs
index 33a6dd6..77be3cb 100644
--- a/lightning-liquidity/tests/lsps2_integration_tests.rs
+++ b/lightning-liquidity/tests/lsps2_integration_tests.rs
@@ -1331,14 +1331,14 @@ fn client_trusts_lsp_end_to_end_test() {
 
 	let total_fee_msat = match service_events[0].clone() {
 		Event::PaymentForwarded {
-			prev_node_id,
-			next_node_id,
+			ref prev_htlcs,
+			ref next_htlcs,
 			skimmed_fee_msat,
 			total_fee_earned_msat,
 			..
 		} => {
-			assert_eq!(prev_node_id, Some(payer_node_id));
-			assert_eq!(next_node_id, Some(client_node_id));
+			assert_eq!(prev_htlcs[0].node_id, Some(payer_node_id));
+			assert_eq!(next_htlcs[0].node_id, Some(client_node_id));
 			service_handler.payment_forwarded(channel_id, skimmed_fee_msat.unwrap_or(0)).unwrap();
 			Some(total_fee_earned_msat.unwrap() - skimmed_fee_msat.unwrap())
 		},
diff --git a/lightning/src/events/mod.rs b/lightning/src/events/mod.rs
index 3f6bb0e..01bbd5d 100644
--- a/lightning/src/events/mod.rs
+++ b/lightning/src/events/mod.rs
@@ -738,6 +738,31 @@ pub enum InboundChannelFunds {
 	DualFunded,
 }
 
+/// Identifies the channel and peer committed to a HTLC, used for both incoming and outgoing HTLCs.
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct HTLCLocator {
+	/// The channel that the HTLC was sent or received on.
+	pub channel_id: ChannelId,
+
+	/// The `user_channel_id` for `channel_id`.
+	///
+	/// This will be `None` if the payment was settled via an on-chain transaction. It will also
+	/// be `None` for events serialized by versions prior to 0.0.122.
+	pub user_channel_id: Option<u128>,
+
+	/// The public key identity of the node that the HTLC was sent to or received from.
+	///
+	/// This is only `None` for HTLCs received prior to 0.1 or for events serialized by versions
+	/// prior to 0.1.
+	pub node_id: Option<PublicKey>,
+}
+
+impl_writeable_tlv_based!(HTLCLocator, {
+	(1, channel_id, required),
+	(3, user_channel_id, option),
+	(5, node_id, option),
+});
+
 /// An Event which you should probably take some action in response to.
 ///
 /// Note that while Writeable and Readable are implemented for Event, you probably shouldn't use
@@ -1331,38 +1356,22 @@ pub enum Event {
 	/// This event is generated when a payment has been successfully forwarded through us and a
 	/// forwarding fee earned.
 	///
+	/// Note that downgrading from 0.3 and above with pending trampoline forwards that use multipart
+	/// payments will produce an event that only provides information about the first htlc that was
+	/// received/dispatched.
+	///
 	/// # Failure Behavior and Persistence
 	/// This event will eventually be replayed after failures-to-handle (i.e., the event handler
 	/// returning `Err(ReplayEvent ())`) and will be persisted across restarts.
 	PaymentForwarded {
-		/// The channel id of the incoming channel between the previous node and us.
-		///
-		/// This is only `None` for events generated or serialized by versions prior to 0.0.107.
-		prev_channel_id: Option<ChannelId>,
-		/// The channel id of the outgoing channel between the next node and us.
-		///
-		/// This is only `None` for events generated or serialized by versions prior to 0.0.107.
-		next_channel_id: Option<ChannelId>,
-		/// The `user_channel_id` of the incoming channel between the previous node and us.
-		///
-		/// This is only `None` for events generated or serialized by versions prior to 0.0.122.
-		prev_user_channel_id: Option<u128>,
-		/// The `user_channel_id` of the outgoing channel between the next node and us.
-		///
-		/// This will be `None` if the payment was settled via an on-chain transaction. See the
-		/// caveat described for the `total_fee_earned_msat` field. Moreover it will be `None` for
-		/// events generated or serialized by versions prior to 0.0.122.
-		next_user_channel_id: Option<u128>,
-		/// The node id of the previous node.
-		///
-		/// This is only `None` for HTLCs received prior to 0.1 or for events serialized by
-		/// versions prior to 0.1
-		prev_node_id: Option<PublicKey>,
-		/// The node id of the next node.
-		///
-		/// This is only `None` for HTLCs received prior to 0.1 or for events serialized by
-		/// versions prior to 0.1
-		next_node_id: Option<PublicKey>,
+		/// The set of HTLCs forwarded to our node that will be claimed by this forward. Contains a
+		/// single HTLC for source-routed payments, and may contain multiple HTLCs when we acted as
+		/// a trampoline router, responsible for pathfinding within the route.
+		prev_htlcs: Vec<HTLCLocator>,
+		/// The set of HTLCs forwarded by our node that have been claimed by this forward. Contains
+		/// a single HTLC for regular source-routed payments, and may contain multiple HTLCs when
+		/// we acted as a trampoline router, responsible for pathfinding within the route.
+		next_htlcs: Vec<HTLCLocator>,
 		/// The total fee, in milli-satoshis, which was earned as a result of the payment.
 		///
 		/// Note that if we force-closed the channel over which we forwarded an HTLC while the HTLC
@@ -2026,29 +2035,47 @@ impl Writeable for Event {
 				});
 			},
 			&Event::PaymentForwarded {
-				prev_channel_id,
-				next_channel_id,
-				prev_user_channel_id,
-				next_user_channel_id,
-				prev_node_id,
-				next_node_id,
+				ref prev_htlcs,
+				ref next_htlcs,
 				total_fee_earned_msat,
 				skimmed_fee_msat,
 				claim_from_onchain_tx,
 				outbound_amount_forwarded_msat,
 			} => {
 				7u8.write(writer)?;
+				// Fields 1, 3, 9, 11, 13 and 15 are written for backwards compatibility. We don't
+				// want to fail writes, so we write garbage data if we don't have at least on htlc.
+				debug_assert!(
+					!prev_htlcs.is_empty(),
+					"at least one prev_htlc required for PaymentForwarded",
+				);
+				debug_assert!(
+					!next_htlcs.is_empty(),
+					"at least one next_htlc required for PaymentForwarded",
+				);
+				let empty_locator = HTLCLocator {
+					channel_id: ChannelId::new_zero(),
+					user_channel_id: None,
+					node_id: None,
+				};
+				let legacy_prev = prev_htlcs.first().unwrap_or(&empty_locator);
+				let legacy_next = next_htlcs.first().unwrap_or(&empty_locator);
 				write_tlv_fields!(writer, {
 					(0, total_fee_earned_msat, option),
-					(1, prev_channel_id, option),
+					(1, Some(legacy_prev.channel_id), option),
 					(2, claim_from_onchain_tx, required),
-					(3, next_channel_id, option),
+					(3, Some(legacy_next.channel_id), option),
 					(5, outbound_amount_forwarded_msat, option),
 					(7, skimmed_fee_msat, option),
-					(9, prev_user_channel_id, option),
-					(11, next_user_channel_id, option),
-					(13, prev_node_id, option),
-					(15, next_node_id, option),
+					(9, legacy_prev.user_channel_id, option),
+					(11, legacy_next.user_channel_id, option),
+					(13, legacy_prev.node_id, option),
+					(15, legacy_next.node_id, option),
+					// HTLCs are written as required, rather than required_vec, so that they can be
+					// deserialized using default_value to fill in legacy fields which expects
+					// LengthReadable (required_vec is WithoutLength).
+					(17, *prev_htlcs, required),
+					(19, *next_htlcs, required),
 				});
 			},
 			&Event::ChannelClosed {
@@ -2548,35 +2575,48 @@ impl MaybeReadable for Event {
 			},
 			7u8 => {
 				let mut f = || {
-					let mut prev_channel_id = None;
-					let mut next_channel_id = None;
-					let mut prev_user_channel_id = None;
-					let mut next_user_channel_id = None;
-					let mut prev_node_id = None;
-					let mut next_node_id = None;
+					// Legacy values that have been replaced by prev_htlcs and next_htlcs.
+					let mut prev_channel_id_legacy = None;
+					let mut next_channel_id_legacy = None;
+					let mut prev_user_channel_id_legacy = None;
+					let mut next_user_channel_id_legacy = None;
+					let mut prev_node_id_legacy = None;
+					let mut next_node_id_legacy = None;
+
 					let mut total_fee_earned_msat = None;
 					let mut skimmed_fee_msat = None;
 					let mut claim_from_onchain_tx = false;
 					let mut outbound_amount_forwarded_msat = None;
+					let mut prev_htlcs = vec![];
+					let mut next_htlcs = vec![];
 					read_tlv_fields!(reader, {
 						(0, total_fee_earned_msat, option),
-						(1, prev_channel_id, option),
+						(1, prev_channel_id_legacy, option),
 						(2, claim_from_onchain_tx, required),
-						(3, next_channel_id, option),
+						(3, next_channel_id_legacy, option),
 						(5, outbound_amount_forwarded_msat, option),
 						(7, skimmed_fee_msat, option),
-						(9, prev_user_channel_id, option),
-						(11, next_user_channel_id, option),
-						(13, prev_node_id, option),
-						(15, next_node_id, option),
+						(9, prev_user_channel_id_legacy, option),
+						(11, next_user_channel_id_legacy, option),
+						(13, prev_node_id_legacy, option),
+						(15, next_node_id_legacy, option),
+						// We never expect prev/next_channel_id_legacy to be None because this field
+						// was only None for versions before 0.0.107 and we do not allow upgrades
+						// with pending forwards to 0.1 for any version 0.0.123 or earlier.
+						(17, prev_htlcs, (default_value, vec![HTLCLocator{
+							channel_id: prev_channel_id_legacy.ok_or(DecodeError::InvalidValue)?,
+							user_channel_id: prev_user_channel_id_legacy,
+							node_id: prev_node_id_legacy,
+						}])),
+						(19, next_htlcs, (default_value, vec![HTLCLocator{
+							channel_id: next_channel_id_legacy.ok_or(DecodeError::InvalidValue)?,
+							user_channel_id: next_user_channel_id_legacy,
+							node_id: next_node_id_legacy,
+						}])),
 					});
 					Ok(Some(Event::PaymentForwarded {
-						prev_channel_id,
-						next_channel_id,
-						prev_user_channel_id,
-						next_user_channel_id,
-						prev_node_id,
-						next_node_id,
+						prev_htlcs,
+						next_htlcs,
 						total_fee_earned_msat,
 						skimmed_fee_msat,
 						claim_from_onchain_tx,
diff --git a/lightning/src/ln/chanmon_update_fail_tests.rs b/lightning/src/ln/chanmon_update_fail_tests.rs
index cd32d21..3642825 100644
--- a/lightning/src/ln/chanmon_update_fail_tests.rs
+++ b/lightning/src/ln/chanmon_update_fail_tests.rs
@@ -3940,11 +3940,11 @@ fn do_test_durable_preimages_on_closed_channel(
 	let evs = nodes[1].node.get_and_clear_pending_events();
 	assert_eq!(evs.len(), if close_chans_before_reload { 2 } else { 1 });
 	for ev in evs {
-		if let Event::PaymentForwarded { claim_from_onchain_tx, next_user_channel_id, .. } = ev {
+		if let Event::PaymentForwarded { claim_from_onchain_tx, next_htlcs, .. } = ev {
 			if !claim_from_onchain_tx {
 				// If the outbound channel is still open, the `next_user_channel_id` should be available.
 				// This was previously broken.
-				assert!(next_user_channel_id.is_some())
+				assert!(next_htlcs[0].user_channel_id.is_some())
 			}
 		} else {
 			panic!();
diff --git a/lightning/src/ln/channelmanager.rs b/lightning/src/ln/channelmanager.rs
index 6bf04cd..a5725a7 100644
--- a/lightning/src/ln/channelmanager.rs
+++ b/lightning/src/ln/channelmanager.rs
@@ -9756,12 +9756,16 @@ This indicates a bug inside LDK. Please report this error at https://github.com/
 							(
 								Some(MonitorUpdateCompletionAction::EmitEventAndFreeOtherChannel {
 									event: events::Event::PaymentForwarded {
-										prev_channel_id: Some(prev_channel_id),
-										next_channel_id: Some(next_channel_id),
-										prev_user_channel_id,
-										next_user_channel_id,
-										prev_node_id,
-										next_node_id: Some(next_channel_counterparty_node_id),
+										prev_htlcs: vec![events::HTLCLocator {
+											channel_id: prev_channel_id,
+											user_channel_id: prev_user_channel_id,
+											node_id: prev_node_id,
+										}],
+										next_htlcs: vec![events::HTLCLocator {
+											channel_id: next_channel_id,
+											user_channel_id: next_user_channel_id,
+											node_id: Some(next_channel_counterparty_node_id),
+										}],
 										total_fee_earned_msat,
 										skimmed_fee_msat,
 										claim_from_onchain_tx: from_onchain,
diff --git a/lightning/src/ln/functional_test_utils.rs b/lightning/src/ln/functional_test_utils.rs
index 2d971c3..641842d 100644
--- a/lightning/src/ln/functional_test_utils.rs
+++ b/lightning/src/ln/functional_test_utils.rs
@@ -3095,17 +3095,16 @@ pub fn expect_payment_forwarded<CM: AChannelManager, H: NodeHolder<CM = CM>>(
 ) -> Option<u64> {
 	match event {
 		Event::PaymentForwarded {
-			prev_channel_id,
-			next_channel_id,
-			prev_user_channel_id,
-			next_user_channel_id,
-			prev_node_id,
-			next_node_id,
+			prev_htlcs,
+			next_htlcs,
 			total_fee_earned_msat,
 			skimmed_fee_msat,
 			claim_from_onchain_tx,
 			..
 		} => {
+			assert_eq!(prev_htlcs.len(), 1);
+			assert_eq!(next_htlcs.len(), 1);
+
 			if allow_1_msat_fee_overpay {
 				// Aggregating fees for blinded paths may result in a rounding error, causing slight
 				// overpayment in fees.
@@ -3120,33 +3119,36 @@ pub fn expect_payment_forwarded<CM: AChannelManager, H: NodeHolder<CM = CM>>(
 			// overpaid amount.
 			assert!(skimmed_fee_msat == expected_extra_fees_msat);
 			if !upstream_force_closed {
-				assert_eq!(prev_node.node().get_our_node_id(), prev_node_id.unwrap());
+				let prev_node_id = prev_htlcs[0].node_id.unwrap();
+				let prev_channel_id = prev_htlcs[0].channel_id;
+				let prev_user_channel_id = prev_htlcs[0].user_channel_id.unwrap();
+
+				assert_eq!(prev_node.node().get_our_node_id(), prev_node_id);
 				// Is the event prev_channel_id in one of the channels between the two nodes?
 				let node_chans = node.node().list_channels();
-				assert!(node_chans.iter().any(|x| x.counterparty.node_id == prev_node_id.unwrap()
-					&& x.channel_id == prev_channel_id.unwrap()
-					&& x.user_channel_id == prev_user_channel_id.unwrap()));
+				assert!(node_chans.iter().any(|x| x.counterparty.node_id == prev_node_id
+					&& x.channel_id == prev_channel_id
+					&& x.user_channel_id == prev_user_channel_id));
 			}
 			// We check for force closures since a force closed channel is removed from the
 			// node's channel list
 			if !downstream_force_closed {
+				let next_node_id = next_htlcs[0].node_id.unwrap();
+				let next_channel_id = next_htlcs[0].channel_id;
+				let next_user_channel_id = next_htlcs[0].user_channel_id.unwrap();
 				// As documented, `next_user_channel_id` will only be `Some` if we didn't settle via an
 				// onchain transaction, just as the `total_fee_earned_msat` field. Rather than
 				// introducing yet another variable, we use the latter's state as a flag to detect
 				// this and only check if it's `Some`.
-				assert_eq!(next_node.node().get_our_node_id(), next_node_id.unwrap());
+				assert_eq!(next_node.node().get_our_node_id(), next_node_id);
 				let node_chans = node.node().list_channels();
 				if total_fee_earned_msat.is_none() {
-					assert!(node_chans
-						.iter()
-						.any(|x| x.counterparty.node_id == next_node_id.unwrap()
-							&& x.channel_id == next_channel_id.unwrap()));
+					assert!(node_chans.iter().any(|x| x.counterparty.node_id == next_node_id
+						&& x.channel_id == next_channel_id));
 				} else {
-					assert!(node_chans
-						.iter()
-						.any(|x| x.counterparty.node_id == next_node_id.unwrap()
-							&& x.channel_id == next_channel_id.unwrap()
-							&& x.user_channel_id == next_user_channel_id.unwrap()));
+					assert!(node_chans.iter().any(|x| x.counterparty.node_id == next_node_id
+						&& x.channel_id == next_channel_id
+						&& x.user_channel_id == next_user_channel_id));
 				}
 			}
 			assert_eq!(claim_from_onchain_tx, downstream_force_closed);
diff --git a/lightning/src/ln/functional_tests.rs b/lightning/src/ln/functional_tests.rs
index 09a87d9..17fbc1f 100644
--- a/lightning/src/ln/functional_tests.rs
+++ b/lightning/src/ln/functional_tests.rs
@@ -1490,37 +1490,37 @@ pub fn test_htlc_on_chain_success() {
 	connect_blocks(&nodes[1], TEST_FINAL_CLTV); // Confirm blocks until the HTLC expires
 	let forwarded_events = nodes[1].node.get_and_clear_pending_events();
 	assert_eq!(forwarded_events.len(), 3);
-	let chan_id = Some(chan_1.2);
+	let chan_id = chan_1.2;
 	match forwarded_events[0] {
 		Event::PaymentForwarded {
+			ref prev_htlcs,
+			ref next_htlcs,
 			total_fee_earned_msat,
-			prev_channel_id,
 			claim_from_onchain_tx,
-			next_channel_id,
 			outbound_amount_forwarded_msat,
 			..
 		} => {
 			assert_eq!(total_fee_earned_msat, Some(1000));
-			assert_eq!(prev_channel_id, chan_id);
+			assert_eq!(prev_htlcs[0].channel_id, chan_id);
 			assert_eq!(claim_from_onchain_tx, true);
-			assert_eq!(next_channel_id, Some(chan_2.2));
+			assert_eq!(next_htlcs[0].channel_id, chan_2.2);
 			assert_eq!(outbound_amount_forwarded_msat, Some(3000000));
 		},
 		_ => panic!(),
 	}
 	match forwarded_events[1] {
 		Event::PaymentForwarded {
+			ref prev_htlcs,
+			ref next_htlcs,
 			total_fee_earned_msat,
-			prev_channel_id,
 			claim_from_onchain_tx,
-			next_channel_id,
 			outbound_amount_forwarded_msat,
 			..
 		} => {
 			assert_eq!(total_fee_earned_msat, Some(1000));
-			assert_eq!(prev_channel_id, chan_id);
+			assert_eq!(prev_htlcs[0].channel_id, chan_id);
 			assert_eq!(claim_from_onchain_tx, true);
-			assert_eq!(next_channel_id, Some(chan_2.2));
+			assert_eq!(next_htlcs[0].channel_id, chan_2.2);
 			assert_eq!(outbound_amount_forwarded_msat, Some(3000000));
 		},
 		_ => panic!(),
@@ -4031,17 +4031,17 @@ pub fn test_onchain_to_onchain_claim() {
 	assert_eq!(events.len(), 2);
 	match events[0] {
 		Event::PaymentForwarded {
+			ref prev_htlcs,
+			ref next_htlcs,
 			total_fee_earned_msat,
-			prev_channel_id,
 			claim_from_onchain_tx,
-			next_channel_id,
 			outbound_amount_forwarded_msat,
 			..
 		} => {
 			assert_eq!(total_fee_earned_msat, Some(1000));
-			assert_eq!(prev_channel_id, Some(chan_1.2));
+			assert_eq!(prev_htlcs[0].channel_id, chan_1.2);
 			assert_eq!(claim_from_onchain_tx, true);
-			assert_eq!(next_channel_id, Some(chan_2.2));
+			assert_eq!(next_htlcs[0].channel_id, chan_2.2);
 			assert_eq!(outbound_amount_forwarded_msat, Some(3000000));
 		},
 		_ => panic!("Unexpected event"),
diff --git a/lightning/src/util/ser.rs b/lightning/src/util/ser.rs
index 2eace55..45ca98b 100644
--- a/lightning/src/util/ser.rs
+++ b/lightning/src/util/ser.rs
@@ -1109,6 +1109,7 @@ impl_for_vec!(crate::routing::router::TrampolineHop);
 impl_for_vec_with_element_length_prefix!(crate::ln::msgs::UpdateAddHTLC);
 impl_writeable_for_vec_with_element_length_prefix!(&crate::ln::msgs::UpdateAddHTLC);
 impl_for_vec!(u32);
+impl_for_vec!(crate::events::HTLCLocator);
 
 impl Writeable for Vec<Witness> {
 	#[inline]