Decode update_add_htlc onions before forwarding HTLCs

This commit completes all of the groundwork necessary to decode incoming
`update_add_htlc` onions once they're fully committed to by both sides.
HTLCs are tracked in batches per-channel on the channel they were
received on. While this path is unreachable for now, until
`InboundHTLCResolution::Resolved` is replaced with
`InboundHTLCResolution::Pending`, it will allow us to obtain
`HTLCHandlingFailed` events for _any_ failed HTLC that comes across a
channel.

Author: wpaulino

lightning/src/ln/channelmanager.rs

@@ -4295,13 +4295,166 @@ where
 		Ok(())
 	}
 
+	pub(crate) fn process_pending_update_add_htlcs(&self) {
+		let mut decode_update_add_htlcs = new_hash_map();
+		mem::swap(&mut decode_update_add_htlcs, &mut self.decode_update_add_htlcs.lock().unwrap());
+
+		let mut incoming_channel_htlc_fails = new_hash_map();
+		let mut htlc_forwards = Vec::new();
+		let mut htlc_fails = Vec::new();
+		for (incoming_scid, update_add_htlcs) in decode_update_add_htlcs {
+			let (incoming_counterparty_node_id, incoming_channel_id) =
+				match self.short_to_chan_info.read().unwrap().get(&incoming_scid).cloned() {
+					Some((cp_id, channel_id)) => (cp_id, channel_id),
+					None => continue,
+				};
+			let per_peer_state = self.per_peer_state.read().unwrap();
+			let peer_state_mutex_opt = per_peer_state.get(&incoming_counterparty_node_id);
+			if peer_state_mutex_opt.is_none() {
+				continue;
+			}
+			let mut peer_state_lock = peer_state_mutex_opt.unwrap().lock().unwrap();
+			let peer_state = &mut *peer_state_lock;
+			let incoming_channel = match peer_state.channel_by_id.get_mut(&incoming_channel_id).map(
+				|chan_phase| if let ChannelPhase::Funded(chan) = chan_phase { Some(chan) } else { None }
+			).flatten() {
+				None => continue,
+				Some(chan) => chan,
+			};
+
+			// Since recursive locks are not allowed, we opt to process all the HTLCs in the
+			// incoming channel first, even though onion failures usually come first. If any of the
+			// HTLCs fail to be accepted, we temporarily track their failure until we decode their
+			// onion below.
+			let incoming_funding_txo = incoming_channel.context.get_funding_txo().unwrap();
+			let incoming_user_channel_id = incoming_channel.context.get_user_id();
+			let incoming_accept_underpaying_htlcs = incoming_channel.context.config().accept_underpaying_htlcs;
+			for update_add_htlc in &update_add_htlcs {
+				let logger = WithChannelContext::from(&self.logger, &incoming_channel.context);
+				if let Err((err, code)) = incoming_channel.can_accept_incoming_htlc(
+					update_add_htlc, &self.fee_estimator, &logger,
+				) {
+					incoming_channel_htlc_fails.insert(update_add_htlc.htlc_id, (err, code));
+				}
+			}
+			mem::drop(peer_state_lock);
+			mem::drop(per_peer_state);
+
+			for update_add_htlc in &update_add_htlcs {
+				// Retrieve the failure for the HTLC on the incoming channel if one exists. Onion
+				// failures and forwarding check failures take precedence over it.
+				let incoming_channel_htlc_fail = incoming_channel_htlc_fails.remove(&update_add_htlc.htlc_id);
+
+				let (next_hop, shared_secret, next_packet_details_opt) = match decode_incoming_update_add_htlc_onion(
+					&update_add_htlc, &self.node_signer, &self.logger, &self.secp_ctx
+				) {
+					Ok(decoded_onion) => decoded_onion,
+					Err(htlc_fail) => {
+						htlc_fails.push((htlc_fail, update_add_htlc.payment_hash, true, None));
+						continue;
+					},
+				};
+
+				let is_intro_node_blinded_forward = next_hop.is_intro_node_blinded_forward();
+				let outgoing_channel_packet_scid_opt = next_packet_details_opt.as_ref().map(|d| d.outgoing_scid);
+
+				// Now process the HTLC on the outgoing channel if it's a forward.
+				let outgoing_channel_update_opt = match next_packet_details_opt.as_ref() {
+					Some(next_packet_details) => match self.can_forward_htlc(
+						&update_add_htlc, &next_packet_details
+					) {
+						Ok(chan_update_opt) => chan_update_opt,
+						Err((err, code, chan_update_opt)) =>  {
+							let htlc_fail = self.htlc_failure_from_update_add_err(
+								&update_add_htlc, &incoming_counterparty_node_id, err, code,
+								chan_update_opt, is_intro_node_blinded_forward, &shared_secret,
+							);
+							htlc_fails.push((htlc_fail, update_add_htlc.payment_hash, false, outgoing_channel_packet_scid_opt));
+							continue;
+						},
+					}
+					None => None,
+				};
+
+				// With the HTLC accepted by the outgoing channel and (potentially) the incoming
+				// channel, we can gather all of the info required internally to forward it.
+				match self.construct_pending_htlc_status(
+					&update_add_htlc, &incoming_counterparty_node_id, shared_secret, next_hop,
+					incoming_accept_underpaying_htlcs, next_packet_details_opt.map(|d| d.next_packet_pubkey),
+				) {
+					PendingHTLCStatus::Forward(htlc_forward) => {
+						// Now that we know we can forward it for sure, check if we had a failure on
+						// the incoming channel.
+						if let Some((err, code)) = incoming_channel_htlc_fail {
+							let htlc_fail = self.htlc_failure_from_update_add_err(
+								&update_add_htlc, &incoming_counterparty_node_id, err, code,
+								outgoing_channel_update_opt, is_intro_node_blinded_forward, &shared_secret,
+							);
+							htlc_fails.push((htlc_fail, update_add_htlc.payment_hash, false, outgoing_channel_packet_scid_opt));
+						} else {
+							htlc_forwards.push((htlc_forward, update_add_htlc.htlc_id));
+						}
+					},
+					PendingHTLCStatus::Fail(htlc_fail) => {
+						htlc_fails.push((htlc_fail, update_add_htlc.payment_hash, false, outgoing_channel_packet_scid_opt));
+					},
+				}
+			}
+			debug_assert!(incoming_channel_htlc_fails.is_empty());
+
+			// Process all of the forwards and failures for the channel in which the HTLCs were
+			// proposed to as a batch.
+			let pending_forwards = (incoming_scid, incoming_funding_txo, incoming_channel_id,
+				incoming_user_channel_id, htlc_forwards.drain(..).collect());
+			self.forward_htlcs_without_forward_event(&mut [pending_forwards]);
+
+			for (htlc_fail, payment_hash, is_invalid_onion, outgoing_scid) in htlc_fails.drain(..) {
+				let failure = match htlc_fail {
+					HTLCFailureMsg::Relay(fail_htlc) => HTLCForwardInfo::FailHTLC {
+						htlc_id: fail_htlc.htlc_id,
+						err_packet: fail_htlc.reason,
+					},
+					HTLCFailureMsg::Malformed(fail_malformed_htlc) => HTLCForwardInfo::FailMalformedHTLC {
+						htlc_id: fail_malformed_htlc.htlc_id,
+						sha256_of_onion: fail_malformed_htlc.sha256_of_onion,
+						failure_code: fail_malformed_htlc.failure_code,
+					},
+				};
+				let destination = if is_invalid_onion {
+					HTLCDestination::InvalidOnion
+				} else if let Some(outgoing_scid) = outgoing_scid {
+					match self.short_to_chan_info.read().unwrap().get(&outgoing_scid) {
+						Some((outgoing_counterparty_node_id, outgoing_channel_id)) =>
+							HTLCDestination::NextHopChannel {
+								node_id: Some(*outgoing_counterparty_node_id),
+								channel_id: *outgoing_channel_id,
+							},
+						None => HTLCDestination::UnknownNextHop {
+							requested_forward_scid: outgoing_scid,
+						},
+					}
+				} else {
+					HTLCDestination::FailedPayment { payment_hash }
+				};
+
+				self.forward_htlcs.lock().unwrap().entry(incoming_scid).or_insert(vec![]).push(failure);
+				self.pending_events.lock().unwrap().push_back((events::Event::HTLCHandlingFailed {
+					prev_channel_id: incoming_channel_id,
+					failed_next_destination: destination,
+				}, None));
+			}
+		}
+	}
+
 	/// Processes HTLCs which are pending waiting on random forward delay.
 	///
 	/// Should only really ever be called in response to a PendingHTLCsForwardable event.
 	/// Will likely generate further events.
 	pub fn process_pending_htlc_forwards(&self) {
 		let _persistence_guard = PersistenceNotifierGuard::notify_on_drop(self);
 
+		self.process_pending_update_add_htlcs();
+
 		let mut new_events = VecDeque::new();
 		let mut failed_forwards = Vec::new();
 		let mut phantom_receives: Vec<(u64, OutPoint, ChannelId, u128, Vec<(PendingHTLCInfo, u64)>)> = Vec::new();