Add a variant to PendingOutboundPayment for retries-exceeded

When a payer gives up trying to retry a payment, they don't know
for sure what the current state of the event queue is.
Specifically, they cannot be sure that there are not multiple
additional `PaymentPathFailed` or even `PaymentSuccess` events
pending which they will see later. Thus, they have a very hard
time identifying whether a payment has truly failed (and informing
the UI of that fact) or if it is still pending. See [1] for more
information.

In order to avoid this mess, we will resolve it here by having the
payer give `ChannelManager` a bit more information - when they
have given up on a payment - and using that to generate a
`PaymentFailed` event when all paths have failed.

This commit adds the neccessary storage and changes for the new
state inside `ChannelManager` and a public method to mark a payment
as failed, the next few commits will add the new `Event` and use
the new features in our `PaymentRetrier`.

[1] lightningdevkit#1164

Author: TheBlueMatt

lightning/src/ln/channelmanager.rs

@@ -454,6 +454,16 @@ pub(crate) enum PendingOutboundPayment {
 		session_privs: HashSet<[u8; 32]>,
 		payment_hash: Option<PaymentHash>,
 	},
+	/// When a payer gives up trying to retry a payment, they inform us, letting us generate a
+	/// `PaymentFailed` event when all HTLCs have irrevocably failed. This avoids a number of race
+	/// conditions in MPP-aware payment retriers[1], where the possibility of multiple
+	/// `PaymentPathFaield` events with `all_paths_failed` can be pending at once, confusing a
+	/// downstream event handler as to when a payment has actually failed.
+	/// [1] https://github.com/lightningdevkit/rust-lightning/issues/1164
+	RetriesExceeded {
+		session_privs: HashSet<[u8; 32]>,
+		payment_hash: PaymentHash,
+	},
 }
 
 impl PendingOutboundPayment {
@@ -469,6 +479,12 @@ impl PendingOutboundPayment {
 			_ => false,
 		}
 	}
+	fn retries_exceeded(&self) -> bool {
+		match self {
+			PendingOutboundPayment::RetriesExceeded { .. } => true,
+			_ => false,
+		}
+	}
 	fn get_pending_fee_msat(&self) -> Option<u64> {
 		match self {
 			PendingOutboundPayment::Retryable { pending_fee_msat, .. } => pending_fee_msat.clone(),
@@ -481,6 +497,7 @@ impl PendingOutboundPayment {
 			PendingOutboundPayment::Legacy { .. } => None,
 			PendingOutboundPayment::Retryable { payment_hash, .. } => Some(*payment_hash),
 			PendingOutboundPayment::Fulfilled { payment_hash, .. } => *payment_hash,
+			PendingOutboundPayment::RetriesExceeded { payment_hash, .. } => Some(*payment_hash),
 		}
 	}
 
@@ -489,19 +506,38 @@ impl PendingOutboundPayment {
 		core::mem::swap(&mut session_privs, match self {
 			PendingOutboundPayment::Legacy { session_privs } |
 			PendingOutboundPayment::Retryable { session_privs, .. } |
-			PendingOutboundPayment::Fulfilled { session_privs, .. }
-				=> session_privs
+			PendingOutboundPayment::Fulfilled { session_privs, .. } |
+			PendingOutboundPayment::RetriesExceeded { session_privs, .. }
+				=> session_privs,
 		});
 		let payment_hash = self.payment_hash();
 		*self = PendingOutboundPayment::Fulfilled { session_privs, payment_hash };
 	}
 
+	fn mark_retries_exceeded(&mut self) -> Result<(), ()> {
+		let mut session_privs = HashSet::new();
+		let our_payment_hash;
+		core::mem::swap(&mut session_privs, match self {
+			PendingOutboundPayment::Legacy { .. } |
+			PendingOutboundPayment::Fulfilled { .. } =>
+				return Err(()),
+			PendingOutboundPayment::Retryable { session_privs, payment_hash, .. } |
+			PendingOutboundPayment::RetriesExceeded { session_privs, payment_hash, .. } => {
+				our_payment_hash = *payment_hash;
+				session_privs
+			},
+		});
+		*self = PendingOutboundPayment::RetriesExceeded { session_privs, payment_hash: our_payment_hash };
+		Ok(())
+	}
+
 	/// panics if path is None and !self.is_fulfilled
 	fn remove(&mut self, session_priv: &[u8; 32], path: Option<&Vec<RouteHop>>) -> bool {
 		let remove_res = match self {
 			PendingOutboundPayment::Legacy { session_privs } |
 			PendingOutboundPayment::Retryable { session_privs, .. } |
-			PendingOutboundPayment::Fulfilled { session_privs, .. } => {
+			PendingOutboundPayment::Fulfilled { session_privs, .. } |
+			PendingOutboundPayment::RetriesExceeded { session_privs, .. } => {
 				session_privs.remove(session_priv)
 			}
 		};
@@ -524,7 +560,8 @@ impl PendingOutboundPayment {
 			PendingOutboundPayment::Retryable { session_privs, .. } => {
 				session_privs.insert(session_priv)
 			}
-			PendingOutboundPayment::Fulfilled { .. } => false
+			PendingOutboundPayment::Fulfilled { .. } => false,
+			PendingOutboundPayment::RetriesExceeded { .. } => false,
 		};
 		if insert_res {
 			if let PendingOutboundPayment::Retryable { ref mut pending_amt_msat, ref mut pending_fee_msat, .. } = self {
@@ -542,7 +579,8 @@ impl PendingOutboundPayment {
 		match self {
 			PendingOutboundPayment::Legacy { session_privs } |
 			PendingOutboundPayment::Retryable { session_privs, .. } |
-			PendingOutboundPayment::Fulfilled { session_privs, .. } => {
+			PendingOutboundPayment::Fulfilled { session_privs, .. } |
+			PendingOutboundPayment::RetriesExceeded { session_privs, .. } => {
 				session_privs.len()
 			}
 		}
@@ -2352,6 +2390,11 @@ impl<Signer: Sign, M: Deref, T: Deref, K: Deref, F: Deref, L: Deref> ChannelMana
 							err: "Payment already completed"
 						}));
 					},
+					PendingOutboundPayment::RetriesExceeded { .. } => {
+						return Err(PaymentSendFailure::ParameterError(APIError::RouteError {
+							err: "Payment already marked timed out (with some HTLCs still pending)"
+						}));
+					},
 				}
 			} else {
 				return Err(PaymentSendFailure::ParameterError(APIError::APIMisuseError {
@@ -2362,6 +2405,14 @@ impl<Signer: Sign, M: Deref, T: Deref, K: Deref, F: Deref, L: Deref> ChannelMana
 		return self.send_payment_internal(route, payment_hash, &payment_secret, None, Some(payment_id), Some(total_msat)).map(|_| ())
 	}
 
+	/// Signals that no further retries for the given payment will occur.
+	pub fn no_further_payment_retries(&self, payment_id: PaymentId) {
+		let mut outbounds = self.pending_outbound_payments.lock().unwrap();
+		if let hash_map::Entry::Occupied(mut payment) = outbounds.entry(payment_id) {
+			let _ = payment.get_mut().mark_retries_exceeded();
+		}
+	}
+
 	/// Send a spontaneous payment, which is a payment that does not require the recipient to have
 	/// generated an invoice. Optionally, you may specify the preimage. If you do choose to specify
 	/// the preimage, it must be a cryptographically secure random value that no intermediate node
@@ -5601,6 +5652,10 @@ impl_writeable_tlv_based_enum_upgradable!(PendingOutboundPayment,
 		(8, pending_amt_msat, required),
 		(10, starting_block_height, required),
 	},
+	(3, RetriesExceeded) => {
+		(0, session_privs, required),
+		(2, payment_hash, required),
+	},
 );
 
 impl<Signer: Sign, M: Deref, T: Deref, K: Deref, F: Deref, L: Deref> Writeable for ChannelManager<Signer, M, T, K, F, L>
@@ -5694,7 +5749,7 @@ impl<Signer: Sign, M: Deref, T: Deref, K: Deref, F: Deref, L: Deref> Writeable f
 		// For backwards compat, write the session privs and their total length.
 		let mut num_pending_outbounds_compat: u64 = 0;
 		for (_, outbound) in pending_outbound_payments.iter() {
-			if !outbound.is_fulfilled() {
+			if !outbound.is_fulfilled() && !outbound.retries_exceeded() {
 				num_pending_outbounds_compat += outbound.remaining_parts() as u64;
 			}
 		}
@@ -5708,6 +5763,7 @@ impl<Signer: Sign, M: Deref, T: Deref, K: Deref, F: Deref, L: Deref> Writeable f
 					}
 				}
 				PendingOutboundPayment::Fulfilled { .. } => {},
+				PendingOutboundPayment::RetriesExceeded { .. } => {},
 			}
 		}