Track currently inflight HTLCs

We add a HashMap to track currently inflight HTLCs as the liquidity
they are using up for each channel.

Author: jurvis

lightning-invoice/src/payment.rs

@@ -145,7 +145,7 @@ use lightning::ln::{PaymentHash, PaymentPreimage, PaymentSecret};
 use lightning::ln::channelmanager::{ChannelDetails, PaymentId, PaymentSendFailure};
 use lightning::ln::msgs::LightningError;
 use lightning::routing::scoring::{LockableScore, Score};
-use lightning::routing::router::{PaymentParameters, Route, RouteParameters};
+use lightning::routing::router::{PaymentParameters, Route, RouteHop, RouteParameters};
 use lightning::util::events::{Event, EventHandler};
 use lightning::util::logger::Logger;
 use time_utils::Time;
@@ -159,6 +159,7 @@ use core::ops::Deref;
 use core::time::Duration;
 #[cfg(feature = "std")]
 use std::time::SystemTime;
+use lightning::util::errors::APIError;
 
 /// A utility for paying [`Invoice`]s and sending spontaneous payments.
 ///
@@ -190,6 +191,7 @@ where
 	/// Caches the overall attempts at making a payment, which is updated prior to retrying.
 	payment_cache: Mutex<HashMap<PaymentHash, PaymentAttempts<T>>>,
 	retry: Retry,
+	total_inflight_map: Mutex<HashMap<(u64, bool), u64>>,
 }
 
 /// Storing minimal payment attempts information required for determining if a outbound payment can
@@ -321,6 +323,7 @@ where
 			event_handler,
 			payment_cache: Mutex::new(HashMap::new()),
 			retry,
+			total_inflight_map: Mutex::new(HashMap::new()),
 		}
 	}
 
@@ -431,7 +434,12 @@ where
 		).map_err(|e| PaymentError::Routing(e))?;
 
 		match send_payment(&route) {
-			Ok(payment_id) => Ok(payment_id),
+			Ok(payment_id) => {
+				for hops in route.paths {
+					self.process_path_inflight_htlcs(&hops);
+				}
+				Ok(payment_id)
+			},
 			Err(e) => match e {
 				PaymentSendFailure::ParameterError(_) => Err(e),
 				PaymentSendFailure::PathParameterError(_) => Err(e),
@@ -446,7 +454,27 @@ where
 						Err(e)
 					}
 				},
-				PaymentSendFailure::PartialFailure { failed_paths_retry, payment_id, .. } => {
+				PaymentSendFailure::PartialFailure { failed_paths_retry, payment_id, results } => {
+					// The ones that are `Ok()`s + `MonitorUpdateFailed` in `results` need to be used to update the map
+					// get the index of them and run to `route` to get the values we need to care about.
+					for (idx, result) in results.iter().enumerate() {
+						match result {
+							Ok(_) => {
+								if let Some(failed_path) = route.paths.get(idx) {
+									self.process_path_inflight_htlcs(failed_path);
+								}
+							}
+							Err(err) => match err {
+								APIError::MonitorUpdateFailed => {
+									if let Some(failed_path) = route.paths.get(idx) {
+										self.process_path_inflight_htlcs(failed_path);
+									}
+								}
+								_ => {}
+							}
+						}
+					}
+
 					if let Some(retry_data) = failed_paths_retry {
 						// Some paths were sent, even if we failed to send the full MPP value our
 						// recipient may misbehave and claim the funds, at which point we have to
@@ -466,6 +494,29 @@ where
 		}.map_err(|e| PaymentError::Sending(e))
 	}
 
+	fn process_path_inflight_htlcs(&self, hops: &Vec<RouteHop>) {
+		// total_inflight_map needs to be direction-sensitive when keeping track of the HTLC value
+		// that is held up. However, the `hops` array, which is a path returned by `find_route` in
+		// the router excludes the payer node. In the following lines, the payer's information is
+		// hardcoded with an inflight value of 0 so that we can correctly represent the first hop
+		// in our sliding window of two.
+		let our_node_id: PublicKey = self.payer.node_id();
+		let reversed_hops_with_payer = core::iter::once((0u64, our_node_id)).chain(
+			hops.split_last().unwrap().1.iter().map(|hop| (hop.fee_msat, hop.pubkey))).rev();
+		let mut cumulative_msat = hops.last().unwrap().fee_msat;
+
+		// Taking the reversed vector from above, we zip it with just the reversed hops list to
+		// work "backwards" of the given path, since the last hop's `fee_msat` actually represents
+		// the total amount sent.
+		for (next_hop, prev_hop) in hops.iter().rev().zip(reversed_hops_with_payer) {
+			cumulative_msat += prev_hop.0;
+			self.total_inflight_map.lock().unwrap()
+				.entry((next_hop.short_channel_id, next_hop.pubkey < prev_hop.1))
+				.and_modify(|used_liquidity_msat| *used_liquidity_msat += cumulative_msat)
+				.or_insert(cumulative_msat);
+		}
+	}
+
 	fn retry_payment(
 		&self, payment_id: PaymentId, payment_hash: PaymentHash, params: &RouteParameters
 	) -> Result<(), ()> {