Deduplicate PendingHTLCsForwardable events when generating

lightning/src/ln/channelmanager.rs

@@ -3748,16 +3748,18 @@ where
 		// being fully configured. See the docs for `ChannelManagerReadArgs` for more.
 		match source {
 			HTLCSource::OutboundRoute { ref path, ref session_priv, ref payment_id, ref payment_params, .. } => {
-				self.pending_outbound_payments.fail_htlc(source, payment_hash, onion_error, path, session_priv, payment_id, payment_params, self.probing_cookie_secret, &self.secp_ctx, &self.pending_events, &self.logger);
+				if self.pending_outbound_payments.fail_htlc(source, payment_hash, onion_error, path,
+					session_priv, payment_id, payment_params, self.probing_cookie_secret, &self.secp_ctx,
+					&self.pending_events, &self.logger).is_some()
+				{ self.push_pending_forwards_ev(); }
 			},
 			HTLCSource::PreviousHopData(HTLCPreviousHopData { ref short_channel_id, ref htlc_id, ref incoming_packet_shared_secret, ref phantom_shared_secret, ref outpoint }) => {
 				log_trace!(self.logger, "Failing HTLC with payment_hash {} backwards from us with {:?}", log_bytes!(payment_hash.0), onion_error);
 				let err_packet = onion_error.get_encrypted_failure_packet(incoming_packet_shared_secret, phantom_shared_secret);
 
-				let mut forward_event = None;
 				let mut forward_htlcs = self.forward_htlcs.lock().unwrap();
 				if forward_htlcs.is_empty() {
-					forward_event = Some(Duration::from_millis(MIN_HTLC_RELAY_HOLDING_CELL_MILLIS));
+					self.push_pending_forwards_ev();
 				}
 				match forward_htlcs.entry(*short_channel_id) {
 					hash_map::Entry::Occupied(mut entry) => {
@@ -3769,11 +3771,6 @@ where
 				}
 				mem::drop(forward_htlcs);
 				let mut pending_events = self.pending_events.lock().unwrap();
-				if let Some(time) = forward_event {
-					pending_events.push(events::Event::PendingHTLCsForwardable {
-						time_forwardable: time
-					});
-				}
 				pending_events.push(events::Event::HTLCHandlingFailed {
 					prev_channel_id: outpoint.to_channel_id(),
 					failed_next_destination: destination,
@@ -3782,6 +3779,19 @@ where
 		}
 	}
 
+	// We only want to push a PendingHTLCsForwardable event if no others are queued.
+	fn push_pending_forwards_ev(&self) {
+		let mut pending_events = self.pending_events.lock().unwrap();
+		let forward_ev_exists = pending_events.iter()
+			.find(|ev| if let events::Event::PendingHTLCsForwardable { .. } = ev { true } else { false })
+			.is_some();
+		if !forward_ev_exists {
+			pending_events.push(events::Event::PendingHTLCsForwardable {
+				time_forwardable: Duration::from_millis(MIN_HTLC_RELAY_HOLDING_CELL_MILLIS),
+			});
+		}
+	}
+
 	/// Provides a payment preimage in response to [`Event::PaymentClaimable`], generating any
 	/// [`MessageSendEvent`]s needed to claim the payment.
 	///
@@ -4908,7 +4918,7 @@ where
 								// We don't want to generate a PendingHTLCsForwardable event if only intercepted
 								// payments are being processed.
 								if forward_htlcs_empty {
-									forward_event = Some(Duration::from_millis(MIN_HTLC_RELAY_HOLDING_CELL_MILLIS));
+									forward_event = Some(());
 								}
 								entry.insert(vec!(HTLCForwardInfo::AddHTLC(PendingAddHTLCInfo {
 									prev_short_channel_id, prev_funding_outpoint, prev_htlc_id, prev_user_channel_id, forward_info })));
@@ -4928,12 +4938,7 @@ where
 			}
 
 			match forward_event {
-				Some(time) => {
-					let mut pending_events = self.pending_events.lock().unwrap();
-					pending_events.push(events::Event::PendingHTLCsForwardable {
-						time_forwardable: time
-					});
-				}
+				Some(()) => self.push_pending_forwards_ev(),
 				None => {},
 			}
 		}

lightning/src/ln/outbound_payment.rs

@@ -15,7 +15,7 @@ use bitcoin::secp256k1::{self, Secp256k1, SecretKey};
 
 use crate::chain::keysinterface::{EntropySource, NodeSigner, Recipient};
 use crate::ln::{PaymentHash, PaymentPreimage, PaymentSecret};
-use crate::ln::channelmanager::{ChannelDetails, HTLCSource, IDEMPOTENCY_TIMEOUT_TICKS, MIN_HTLC_RELAY_HOLDING_CELL_MILLIS, PaymentId};
+use crate::ln::channelmanager::{ChannelDetails, HTLCSource, IDEMPOTENCY_TIMEOUT_TICKS, PaymentId};
 use crate::ln::channelmanager::MIN_FINAL_CLTV_EXPIRY_DELTA as LDK_DEFAULT_MIN_FINAL_CLTV_EXPIRY_DELTA;
 use crate::ln::msgs::DecodeError;
 use crate::ln::onion_utils::HTLCFailReason;
@@ -30,7 +30,6 @@ use crate::util::time::tests::SinceEpoch;
 use core::cmp;
 use core::fmt::{self, Display, Formatter};
 use core::ops::Deref;
-use core::time::Duration;
 
 use crate::prelude::*;
 use crate::sync::Mutex;
@@ -990,12 +989,13 @@ impl OutboundPayments {
 		});
 	}
 
+	// Returns a bool indicating whether a PendingHTLCsForwardable event should be generated.
 	pub(super) fn fail_htlc<L: Deref>(
 		&self, source: &HTLCSource, payment_hash: &PaymentHash, onion_error: &HTLCFailReason,
 		path: &Vec<RouteHop>, session_priv: &SecretKey, payment_id: &PaymentId,
 		payment_params: &Option<PaymentParameters>, probing_cookie_secret: [u8; 32],
 		secp_ctx: &Secp256k1<secp256k1::All>, pending_events: &Mutex<Vec<events::Event>>, logger: &L
-	) where L::Target: Logger {
+	) -> Option<()> where L::Target: Logger {
 		#[cfg(test)]
 		let (network_update, short_channel_id, payment_retryable, onion_error_code, onion_error_data) = onion_error.decode_onion_failure(secp_ctx, logger, &source);
 		#[cfg(not(test))]
@@ -1011,11 +1011,11 @@ impl OutboundPayments {
 		let attempts_remaining = if let hash_map::Entry::Occupied(mut payment) = outbounds.entry(*payment_id) {
 			if !payment.get_mut().remove(&session_priv_bytes, Some(&path)) {
 				log_trace!(logger, "Received duplicative fail for HTLC with payment_hash {}", log_bytes!(payment_hash.0));
-				return
+				return None
 			}
 			if payment.get().is_fulfilled() {
 				log_trace!(logger, "Received failure of HTLC with payment_hash {} after payment completion", log_bytes!(payment_hash.0));
-				return
+				return None
 			}
 			let is_retryable_now = payment.get().is_auto_retryable_now();
 			if let Some(scid) = short_channel_id {
@@ -1061,7 +1061,7 @@ impl OutboundPayments {
 			is_retryable_now
 		} else {
 			log_trace!(logger, "Received duplicative fail for HTLC with payment_hash {}", log_bytes!(payment_hash.0));
-			return
+			return None
 		};
 		core::mem::drop(outbounds);
 		log_trace!(logger, "Failing outbound payment HTLC with payment_hash {}", log_bytes!(payment_hash.0));
@@ -1091,9 +1091,7 @@ impl OutboundPayments {
 				}
 				if payment_retryable && attempts_remaining && retry.is_some() {
 					debug_assert!(full_failure_ev.is_none());
-					pending_retry_ev = Some(events::Event::PendingHTLCsForwardable {
-						time_forwardable: Duration::from_millis(MIN_HTLC_RELAY_HOLDING_CELL_MILLIS),
-					});
+					pending_retry_ev = Some(());
 				}
 				events::Event::PaymentPathFailed {
 					payment_id: Some(*payment_id),
@@ -1114,7 +1112,7 @@ impl OutboundPayments {
 		let mut pending_events = pending_events.lock().unwrap();
 		pending_events.push(path_failure);
 		if let Some(ev) = full_failure_ev { pending_events.push(ev); }
-		if let Some(ev) = pending_retry_ev { pending_events.push(ev); }
+		pending_retry_ev
 	}
 
 	pub(super) fn abandon_payment(&self, payment_id: PaymentId) -> Option<events::Event> {

lightning/src/ln/payment_tests.rs

@@ -2404,7 +2404,7 @@ fn no_extra_retries_on_back_to_back_fail() {
 	// Because we now retry payments as a batch, we simply return a single-path route in the
 	// second, batched, request, have that fail, then complete the payment via `abandon_payment`.
 	let mut events = nodes[0].node.get_and_clear_pending_events();
-	assert_eq!(events.len(), 4);
+	assert_eq!(events.len(), 3);
 	match events[0] {
 		Event::PaymentPathFailed { payment_hash: ev_payment_hash, payment_failed_permanently, ..  } => {
 			assert_eq!(payment_hash, ev_payment_hash);
@@ -2423,10 +2423,6 @@ fn no_extra_retries_on_back_to_back_fail() {
 		},
 		_ => panic!("Unexpected event"),
 	}
-	match events[3] {
-		Event::PendingHTLCsForwardable { .. } => {},
-		_ => panic!("Unexpected event"),
-	}
 
 	nodes[0].node.process_pending_htlc_forwards();
 	let retry_htlc_updates = SendEvent::from_node(&nodes[0]);