f: store HTLCPreviousHopData in Vec for TrampolineForward HTLCSource (todo: split up commit)

Author: arik-so

lightning/src/chain/channelmonitor.rs

@@ -3173,7 +3173,7 @@ impl<Signer: EcdsaChannelSigner> ChannelMonitorImpl<Signer> {
 					} else { false }
 				}));
 			}
-			self.counterparty_fulfilled_htlcs.insert(*claimed_htlc_id, *claimed_preimage);
+			self.counterparty_fulfilled_htlcs.insert(claimed_htlc_id.clone(), *claimed_preimage);
 		}
 	}
 

lightning/src/ln/channelmanager.rs

@@ -623,11 +623,17 @@ impl Readable for InterceptId {
 }
 
 #[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
+pub(crate) struct PreviousHopIdData {
+	short_channel_id: u64,
+	htlc_id: u64,
+}
+
+#[derive(Clone, Debug, PartialEq, Eq, Hash)]
 /// Uniquely describes an HTLC by its source. Just the guaranteed-unique subset of [`HTLCSource`].
 pub(crate) enum SentHTLCId {
 	PreviousHopData { short_channel_id: u64, htlc_id: u64 },
 	OutboundRoute { session_priv: [u8; SECRET_KEY_SIZE] },
-	TrampolineForward { session_priv: [u8; SECRET_KEY_SIZE], previous_short_channel_id: u64, htlc_id: u64 }
+	TrampolineForward { session_priv: [u8; SECRET_KEY_SIZE], previous_hop_data: Vec<PreviousHopIdData> }
 }
 impl SentHTLCId {
 	pub(crate) fn from_source(source: &HTLCSource) -> Self {
@@ -640,12 +646,18 @@ impl SentHTLCId {
 				Self::OutboundRoute { session_priv: session_priv.secret_bytes() },
 			HTLCSource::TrampolineForward { previous_hop_data, session_priv, .. } => Self::TrampolineForward {
 				session_priv: session_priv.secret_bytes(),
-				previous_short_channel_id: previous_hop_data.short_channel_id,
-				htlc_id: previous_hop_data.htlc_id,
+				previous_hop_data: previous_hop_data.iter().map(|hop_data| PreviousHopIdData {
+					short_channel_id: hop_data.short_channel_id,
+					htlc_id: hop_data.htlc_id,
+				}).collect(),
 			},
 		}
 	}
 }
+impl_writeable_tlv_based!(PreviousHopIdData, {
+	(0, short_channel_id, required),
+	(2, htlc_id, required),
+});
 impl_writeable_tlv_based_enum!(SentHTLCId,
 	(0, PreviousHopData) => {
 		(0, short_channel_id, required),
@@ -656,8 +668,7 @@ impl_writeable_tlv_based_enum!(SentHTLCId,
 	},
 	(4, TrampolineForward) => {
 		(0, session_priv, required),
-		(2, previous_short_channel_id, required),
-		(4, htlc_id, required),
+		(2, previous_hop_data, required_vec),
 	},
 );
 
@@ -670,7 +681,8 @@ mod fuzzy_channelmanager {
 	pub enum HTLCSource {
 		PreviousHopData(HTLCPreviousHopData),
 		TrampolineForward {
-			previous_hop_data: HTLCPreviousHopData,
+			// we might be forwarding an incoming payment that was received over MPP
+			previous_hop_data: Vec<HTLCPreviousHopData>,
 			incoming_trampoline_shared_secret: [u8; 32],
 			hops: Vec<RouteHop>,
 			session_priv: SecretKey,
@@ -6231,7 +6243,7 @@ where
 								let htlc_source = HTLCSource::TrampolineForward {
 									// dummy value
 									session_priv: SecretKey::from_slice(&self.entropy_source.get_secure_random_bytes()).unwrap(),
-									previous_hop_data: HTLCPreviousHopData {
+									previous_hop_data: vec![HTLCPreviousHopData {
 										short_channel_id: prev_short_channel_id,
 										user_channel_id: Some(prev_user_channel_id),
 										counterparty_node_id: prev_counterparty_node_id,
@@ -6243,7 +6255,7 @@ where
 										phantom_shared_secret: None,
 										blinded_failure: blinded.map(|b| b.failure),
 										cltv_expiry: Some(incoming_cltv_expiry),
-									},
+									}],
 									incoming_trampoline_shared_secret,
 									hops: Vec::new(),
 								};
@@ -7303,7 +7315,7 @@ where
 		// Note that we MUST NOT end up calling methods on self.chain_monitor here - we're called
 		// from block_connected which may run during initialization prior to the chain_monitor
 		// being fully configured. See the docs for `ChannelManagerReadArgs` for more.
-		let mut push_forward_event;
+		let mut push_forward_event = true;
 		match source {
 			HTLCSource::OutboundRoute { ref path, ref session_priv, ref payment_id, .. } => {
 				push_forward_event = self.pending_outbound_payments.fail_htlc(source, payment_hash, onion_error, path,
@@ -7364,56 +7376,61 @@ where
 				// todo: what do we want to do with this given we do not wish to propagate it directly?
 				let _decoded_onion_failure = onion_error.decode_onion_failure(&self.secp_ctx, &self.logger, &source);
 
-				let incoming_packet_shared_secret = previous_hop_data.incoming_packet_shared_secret;
-				let channel_id = previous_hop_data.channel_id;
-				let short_channel_id = previous_hop_data.short_channel_id;
-				let htlc_id = previous_hop_data.htlc_id;
-				let blinded_failure = previous_hop_data.blinded_failure;
-				log_trace!(
-					WithContext::from(&self.logger, None, Some(channel_id), Some(*payment_hash)),
-					"Failing {}HTLC with payment_hash {} backwards from us following Trampoline forwarding failure: {:?}",
-					if blinded_failure.is_some() { "blinded " } else { "" }, &payment_hash, onion_error
-				);
-				let failure = match blinded_failure {
-					Some(BlindedFailure::FromIntroductionNode) => {
-						let blinded_onion_error = HTLCFailReason::reason(INVALID_ONION_BLINDING, vec![0; 32]);
-						let err_packet = blinded_onion_error.get_encrypted_failure_packet(
-							&incoming_packet_shared_secret, &Some(incoming_trampoline_shared_secret.clone())
-						);
-						HTLCForwardInfo::FailHTLC { htlc_id, err_packet }
-					},
-					Some(BlindedFailure::FromBlindedNode) => {
-						HTLCForwardInfo::FailMalformedHTLC {
-							htlc_id,
-							failure_code: INVALID_ONION_BLINDING,
-							sha256_of_onion: [0; 32]
+				for current_hop_data in previous_hop_data {
+					let incoming_packet_shared_secret = current_hop_data.incoming_packet_shared_secret;
+					let channel_id = current_hop_data.channel_id;
+					let short_channel_id = current_hop_data.short_channel_id;
+					let htlc_id = current_hop_data.htlc_id;
+					let blinded_failure = current_hop_data.blinded_failure;
+					log_trace!(
+						WithContext::from(&self.logger, None, Some(channel_id), Some(*payment_hash)),
+						"Failing {}HTLC with payment_hash {} backwards from us following Trampoline forwarding failure: {:?}",
+						if blinded_failure.is_some() { "blinded " } else { "" }, &payment_hash, onion_error
+					);
+					let failure = match blinded_failure {
+						Some(BlindedFailure::FromIntroductionNode) => {
+							let blinded_onion_error = HTLCFailReason::reason(INVALID_ONION_BLINDING, vec![0; 32]);
+							let err_packet = blinded_onion_error.get_encrypted_failure_packet(
+								&incoming_packet_shared_secret, &Some(incoming_trampoline_shared_secret.clone())
+							);
+							HTLCForwardInfo::FailHTLC { htlc_id, err_packet }
+						},
+						Some(BlindedFailure::FromBlindedNode) => {
+							HTLCForwardInfo::FailMalformedHTLC {
+								htlc_id,
+								failure_code: INVALID_ONION_BLINDING,
+								sha256_of_onion: [0; 32]
+							}
+						},
+						None => {
+							let err_code = 0x2000 | 25;
+							let err_packet = HTLCFailReason::reason(err_code, Vec::new())
+								.get_encrypted_failure_packet(&incoming_packet_shared_secret, &Some(incoming_trampoline_shared_secret.clone()));
+							HTLCForwardInfo::FailHTLC { htlc_id, err_packet }
 						}
-					},
-					None => {
-						let err_code = 0x2000 | 25;
-						let err_packet = HTLCFailReason::reason(err_code, Vec::new())
-							.get_encrypted_failure_packet(&incoming_packet_shared_secret, &Some(incoming_trampoline_shared_secret.clone()));
-						HTLCForwardInfo::FailHTLC { htlc_id, err_packet }
-					}
-				};
+					};
 
-				push_forward_event = self.decode_update_add_htlcs.lock().unwrap().is_empty();
-				let mut forward_htlcs = self.forward_htlcs.lock().unwrap();
-				push_forward_event &= forward_htlcs.is_empty();
-				match forward_htlcs.entry(short_channel_id) {
-					hash_map::Entry::Occupied(mut entry) => {
-						entry.get_mut().push(failure);
-					},
-					hash_map::Entry::Vacant(entry) => {
-						entry.insert(vec!(failure));
+					push_forward_event = self.decode_update_add_htlcs.lock().unwrap().is_empty();
+					let mut forward_htlcs = self.forward_htlcs.lock().unwrap();
+					push_forward_event &= forward_htlcs.is_empty();
+
+					match forward_htlcs.entry(short_channel_id) {
+						hash_map::Entry::Occupied(mut entry) => {
+							entry.get_mut().push(failure);
+						},
+						hash_map::Entry::Vacant(entry) => {
+							entry.insert(vec!(failure));
+						}
 					}
+
+					mem::drop(forward_htlcs);
+
+					let mut pending_events = self.pending_events.lock().unwrap();
+					pending_events.push_back((events::Event::HTLCHandlingFailed {
+						prev_channel_id: channel_id,
+						failed_next_destination: destination.clone(),
+					}, None));
 				}
-				mem::drop(forward_htlcs);
-				let mut pending_events = self.pending_events.lock().unwrap();
-				pending_events.push_back((events::Event::HTLCHandlingFailed {
-					prev_channel_id: channel_id,
-					failed_next_destination: destination,
-				}, None));
 			},
 		}
 		push_forward_event
@@ -7819,7 +7836,7 @@ This indicates a bug inside LDK. Please report this error at https://github.com/
 					session_priv, path, from_onchain, ev_completion_action, &self.pending_events,
 					&self.logger);
 			},
-			HTLCSource::PreviousHopData(hop_data) | HTLCSource::TrampolineForward { previous_hop_data: hop_data, .. } => {
+			HTLCSource::PreviousHopData(hop_data) => {
 				let prev_channel_id = hop_data.channel_id;
 				let prev_user_channel_id = hop_data.user_channel_id;
 				let prev_node_id = hop_data.counterparty_node_id;
@@ -7874,6 +7891,63 @@ This indicates a bug inside LDK. Please report this error at https://github.com/
 						}
 					});
 			},
+			HTLCSource::TrampolineForward { previous_hop_data, .. } => {
+				for current_previous_hop_data in previous_hop_data {
+					let prev_channel_id = current_previous_hop_data.channel_id;
+					let prev_user_channel_id = current_previous_hop_data.user_channel_id;
+					let prev_node_id = current_previous_hop_data.counterparty_node_id;
+					let completed_blocker = RAAMonitorUpdateBlockingAction::from_prev_hop_data(&current_previous_hop_data);
+					self.claim_funds_from_hop(current_previous_hop_data, payment_preimage, None,
+						|htlc_claim_value_msat, definitely_duplicate| {
+							let chan_to_release = Some(EventUnblockedChannel {
+								counterparty_node_id: next_channel_counterparty_node_id,
+								funding_txo: next_channel_outpoint,
+								channel_id: next_channel_id,
+								blocking_action: completed_blocker,
+							});
+
+							if definitely_duplicate && startup_replay {
+								// On startup we may get redundant claims which are related to
+								// monitor updates still in flight. In that case, we shouldn't
+								// immediately free, but instead let that monitor update complete
+								// in the background.
+								(None, None)
+							} else if definitely_duplicate {
+								if let Some(other_chan) = chan_to_release {
+									(Some(MonitorUpdateCompletionAction::FreeOtherChannelImmediately {
+										downstream_counterparty_node_id: other_chan.counterparty_node_id,
+										downstream_funding_outpoint: other_chan.funding_txo,
+										downstream_channel_id: other_chan.channel_id,
+										blocking_action: other_chan.blocking_action,
+									}), None)
+								} else { (None, None) }
+							} else {
+								let total_fee_earned_msat = if let Some(forwarded_htlc_value) = forwarded_htlc_value_msat {
+									if let Some(claimed_htlc_value) = htlc_claim_value_msat {
+										Some(claimed_htlc_value - forwarded_htlc_value)
+									} else { None }
+								} else { None };
+								debug_assert!(skimmed_fee_msat <= total_fee_earned_msat,
+									"skimmed_fee_msat must always be included in total_fee_earned_msat");
+								(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),
+										total_fee_earned_msat,
+										skimmed_fee_msat,
+										claim_from_onchain_tx: from_onchain,
+										outbound_amount_forwarded_msat: forwarded_htlc_value_msat,
+									},
+									downstream_counterparty_and_funding_outpoint: chan_to_release,
+								}), None)
+							}
+						});
+				}
+			},
 		}
 	}
 
@@ -13525,18 +13599,18 @@ impl Readable for HTLCSource {
 			}
 			1 => Ok(HTLCSource::PreviousHopData(Readable::read(reader)?)),
 			2 => {
-				let mut previous_hop_data: crate::util::ser::RequiredWrapper<HTLCPreviousHopData> = crate::util::ser::RequiredWrapper(None);
+				let mut previous_hop_data = Vec::new();
 				let mut incoming_trampoline_shared_secret: crate::util::ser::RequiredWrapper<[u8; 32]> = crate::util::ser::RequiredWrapper(None);
 				let mut session_priv: crate::util::ser::RequiredWrapper<SecretKey> = crate::util::ser::RequiredWrapper(None);
 				let mut hops = Vec::new();
 				read_tlv_fields!(reader, {
-					(0, previous_hop_data, required),
+					(0, previous_hop_data, required_vec),
 					(2, incoming_trampoline_shared_secret, required),
 					(4, session_priv, required),
 					(6, hops, required_vec),
 				});
 				Ok(HTLCSource::TrampolineForward {
-					previous_hop_data: previous_hop_data.0.unwrap(),
+					previous_hop_data,
 					incoming_trampoline_shared_secret: incoming_trampoline_shared_secret.0.unwrap(),
 					hops,
 					session_priv: session_priv.0.unwrap(),
@@ -13567,11 +13641,12 @@ impl Writeable for HTLCSource {
 				1u8.write(writer)?;
 				field.write(writer)?;
 			}
-			HTLCSource::TrampolineForward { ref previous_hop_data, ref incoming_trampoline_shared_secret, ref session_priv, hops: hops_ref } => {
+			HTLCSource::TrampolineForward { previous_hop_data: previous_hop_data_ref, ref incoming_trampoline_shared_secret, ref session_priv, hops: hops_ref } => {
 				2u8.write(writer)?;
+				let previous_hop_data = previous_hop_data_ref.clone();
 				let hops = hops_ref.clone();
 				write_tlv_fields!(writer, {
-					(0, previous_hop_data, required),
+					(0, previous_hop_data, required_vec),
 					(2, incoming_trampoline_shared_secret, required),
 					(4, session_priv, required),
 					(6, hops, required_vec),
@@ -14710,7 +14785,7 @@ where
 					for (htlc_source, (htlc, preimage_opt)) in monitor.get_all_current_outbound_htlcs() {
 						let logger = WithChannelMonitor::from(&args.logger, monitor, Some(htlc.payment_hash));
 						match htlc_source {
-							HTLCSource::PreviousHopData(prev_hop_data) | HTLCSource::TrampolineForward { previous_hop_data: prev_hop_data, .. } => {
+							HTLCSource::PreviousHopData(prev_hop_data) => {
 								let pending_forward_matches_htlc = |info: &PendingAddHTLCInfo| {
 									info.prev_funding_outpoint == prev_hop_data.outpoint &&
 										info.prev_htlc_id == prev_hop_data.htlc_id
@@ -14757,6 +14832,55 @@ where
 									} else { true }
 								});
 							},
+							HTLCSource::TrampolineForward { previous_hop_data, .. } => {
+								for current_previous_hop_data in previous_hop_data {
+									let pending_forward_matches_htlc = |info: &PendingAddHTLCInfo| {
+										info.prev_funding_outpoint == current_previous_hop_data.outpoint &&
+											info.prev_htlc_id == current_previous_hop_data.htlc_id
+									};
+									// 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` or
+									// `pending_intercepted_htlcs`, we were apparently not persisted after
+									// the monitor was when forwarding the payment.
+									decode_update_add_htlcs.retain(|scid, update_add_htlcs| {
+										update_add_htlcs.retain(|update_add_htlc| {
+											let matches = *scid == current_previous_hop_data.short_channel_id &&
+												update_add_htlc.htlc_id == current_previous_hop_data.htlc_id;
+											if matches {
+												log_info!(logger, "Removing pending to-decode HTLC with hash {} as it was forwarded to the closed channel {}",
+												&htlc.payment_hash, &monitor.channel_id());
+											}
+											!matches
+										});
+										!update_add_htlcs.is_empty()
+									});
+									forward_htlcs.retain(|_, forwards| {
+										forwards.retain(|forward| {
+											if let HTLCForwardInfo::AddHTLC(htlc_info) = forward {
+												if pending_forward_matches_htlc(&htlc_info) {
+													log_info!(logger, "Removing pending to-forward HTLC with hash {} as it was forwarded to the closed channel {}",
+													&htlc.payment_hash, &monitor.channel_id());
+													false
+												} else { true }
+											} else { true }
+										});
+										!forwards.is_empty()
+									});
+									pending_intercepted_htlcs.as_mut().unwrap().retain(|intercepted_id, htlc_info| {
+										if pending_forward_matches_htlc(&htlc_info) {
+											log_info!(logger, "Removing pending intercepted HTLC with hash {} as it was forwarded to the closed channel {}",
+											&htlc.payment_hash, &monitor.channel_id());
+											pending_events_read.retain(|(event, _)| {
+												if let Event::HTLCIntercepted { intercept_id: ev_id, .. } = event {
+													intercepted_id != ev_id
+												} else { true }
+											});
+											false
+										} else { true }
+									});
+								}
+							}
 							HTLCSource::OutboundRoute { payment_id, session_priv, path, .. } => {
 								if let Some(preimage) = preimage_opt {
 									let pending_events = Mutex::new(pending_events_read);

lightning/src/util/ser.rs

@@ -1075,6 +1075,7 @@ impl Readable for Vec<u8> {
 
 impl_for_vec!(ecdsa::Signature);
 impl_for_vec!(crate::chain::channelmonitor::ChannelMonitorUpdate);
+impl_for_vec!(crate::ln::channelmanager::HTLCPreviousHopData);
 impl_for_vec!(crate::ln::channelmanager::MonitorUpdateCompletionAction);
 impl_for_vec!(crate::ln::channelmanager::PaymentClaimDetails);
 impl_for_vec!(crate::ln::msgs::SocketAddress);