feat: track in-flight amount for pending payments

Tracks actual in-flight amounts for pending payments to enable accurate balance calculations in wallets

- Added payment_id to Balance::MaybeTimeoutClaimableHTLC so monitors can report which payment each HTLC belongs to
- Added is_retryable to RecentPaymentDetails::Pending to distinguish active retries from stuck HTLCs
- Added reconcile_inflight_payments() using ChannelMonitor as authoritative source for amounts to prevent double-counting
- Added aggregate_outbound_htlcs_by_payment() to ChannelMonitor for aggregating HTLCs by payment

This solves the race condition when fetching from ChannelMonitor and ChannelManager separately during retries

Author: shruti2522

lightning/src/chain/balance_aggregation_tests.rs

@@ -0,0 +1,150 @@
+// This file is Copyright its original authors, visible in version control
+// history.
+//
+// This file is licensed under the Apache License, Version 2.0 <LICENSE-APACHE
+// or http://www.apache.org/licenses/LICENSE-2.0> or the MIT license
+// <LICENSE-MIT or http://opensource.org/licenses/MIT>, at your option.
+// You may not use this file except in accordance with one or both of these
+// licenses.
+
+use super::channelmonitor::Balance;
+use crate::ln::channelmanager::PaymentId;
+use crate::types::payment::PaymentHash;
+
+#[test]
+fn test_aggregate_outbound_htlcs_by_payment() {
+	// Empty balances
+	let balances = vec![];
+	let aggregated = Balance::aggregate_outbound_htlcs_by_payment(&balances);
+	assert!(aggregated.is_empty());
+
+	// Single payment
+	let payment_id_1 = PaymentId([1; 32]);
+	let balances = vec![Balance::MaybeTimeoutClaimableHTLC {
+		amount_satoshis: 10_000,
+		claimable_height: 100,
+		payment_hash: PaymentHash([0; 32]),
+		payment_id: Some(payment_id_1),
+		outbound_payment: true,
+	}];
+	let aggregated = Balance::aggregate_outbound_htlcs_by_payment(&balances);
+	assert_eq!(aggregated.len(), 1);
+	assert_eq!(*aggregated.get(&payment_id_1).unwrap(), 10_000);
+
+	// Test multiple HTLCs for same payment
+	let balances = vec![
+		Balance::MaybeTimeoutClaimableHTLC {
+			amount_satoshis: 10_000,
+			claimable_height: 100,
+			payment_hash: PaymentHash([0; 32]),
+			payment_id: Some(payment_id_1),
+			outbound_payment: true,
+		},
+		Balance::MaybeTimeoutClaimableHTLC {
+			amount_satoshis: 15_000,
+			claimable_height: 100,
+			payment_hash: PaymentHash([0; 32]),
+			payment_id: Some(payment_id_1),
+			outbound_payment: true,
+		},
+		Balance::MaybeTimeoutClaimableHTLC {
+			amount_satoshis: 5_000,
+			claimable_height: 100,
+			payment_hash: PaymentHash([0; 32]),
+			payment_id: Some(payment_id_1),
+			outbound_payment: true,
+		},
+	];
+	let aggregated = Balance::aggregate_outbound_htlcs_by_payment(&balances);
+	assert_eq!(aggregated.len(), 1);
+	assert_eq!(*aggregated.get(&payment_id_1).unwrap(), 30_000);
+
+	// Test multiple distinct payments
+	let payment_id_2 = PaymentId([2; 32]);
+	let balances = vec![
+		Balance::MaybeTimeoutClaimableHTLC {
+			amount_satoshis: 10_000,
+			claimable_height: 100,
+			payment_hash: PaymentHash([0; 32]),
+			payment_id: Some(payment_id_1),
+			outbound_payment: true,
+		},
+		Balance::MaybeTimeoutClaimableHTLC {
+			amount_satoshis: 20_000,
+			claimable_height: 100,
+			payment_hash: PaymentHash([1; 32]),
+			payment_id: Some(payment_id_2),
+			outbound_payment: true,
+		},
+	];
+	let aggregated = Balance::aggregate_outbound_htlcs_by_payment(&balances);
+	assert_eq!(aggregated.len(), 2);
+	assert_eq!(*aggregated.get(&payment_id_1).unwrap(), 10_000);
+	assert_eq!(*aggregated.get(&payment_id_2).unwrap(), 20_000);
+
+	// Should ignore forwarded HTLCs
+	let balances = vec![
+		Balance::MaybeTimeoutClaimableHTLC {
+			amount_satoshis: 10_000,
+			claimable_height: 100,
+			payment_hash: PaymentHash([0; 32]),
+			payment_id: Some(payment_id_1),
+			outbound_payment: true,
+		},
+		Balance::MaybeTimeoutClaimableHTLC {
+			amount_satoshis: 50_000,
+			claimable_height: 100,
+			payment_hash: PaymentHash([1; 32]),
+			payment_id: None,
+			outbound_payment: false,
+		},
+	];
+	let aggregated = Balance::aggregate_outbound_htlcs_by_payment(&balances);
+	assert_eq!(aggregated.len(), 1);
+	assert_eq!(*aggregated.get(&payment_id_1).unwrap(), 10_000);
+
+	// Should ignore outbound HTLCs without payment_id
+	let balances = vec![
+		Balance::MaybeTimeoutClaimableHTLC {
+			amount_satoshis: 10_000,
+			claimable_height: 100,
+			payment_hash: PaymentHash([0; 32]),
+			payment_id: Some(payment_id_1),
+			outbound_payment: true,
+		},
+		Balance::MaybeTimeoutClaimableHTLC {
+			amount_satoshis: 99_000,
+			claimable_height: 100,
+			payment_hash: PaymentHash([2; 32]),
+			payment_id: None,
+			outbound_payment: true,
+		},
+	];
+	let aggregated = Balance::aggregate_outbound_htlcs_by_payment(&balances);
+	assert_eq!(aggregated.len(), 1);
+	assert_eq!(*aggregated.get(&payment_id_1).unwrap(), 10_000);
+
+	// Should ignore other balance types
+	let balances = vec![
+		Balance::MaybeTimeoutClaimableHTLC {
+			amount_satoshis: 10_000,
+			claimable_height: 100,
+			payment_hash: PaymentHash([0; 32]),
+			payment_id: Some(payment_id_1),
+			outbound_payment: true,
+		},
+		Balance::MaybePreimageClaimableHTLC {
+			amount_satoshis: 50_000,
+			expiry_height: 100,
+			payment_hash: PaymentHash([1; 32]),
+		},
+		Balance::ClaimableAwaitingConfirmations {
+			amount_satoshis: 100_000,
+			confirmation_height: 100,
+			source: crate::chain::channelmonitor::BalanceSource::Htlc,
+		},
+	];
+	let aggregated = Balance::aggregate_outbound_htlcs_by_payment(&balances);
+	assert_eq!(aggregated.len(), 1);
+	assert_eq!(*aggregated.get(&payment_id_1).unwrap(), 10_000);
+}

lightning/src/chain/channelmonitor.rs

@@ -55,7 +55,7 @@ use crate::ln::channel_keys::{
 	DelayedPaymentBasepoint, DelayedPaymentKey, HtlcBasepoint, HtlcKey, RevocationBasepoint,
 	RevocationKey,
 };
-use crate::ln::channelmanager::{HTLCSource, PaymentClaimDetails, SentHTLCId};
+use crate::ln::channelmanager::{HTLCSource, PaymentClaimDetails, PaymentId, SentHTLCId};
 use crate::ln::msgs::DecodeError;
 use crate::ln::types::ChannelId;
 use crate::sign::{
@@ -926,6 +926,9 @@ pub enum Balance {
 		claimable_height: u32,
 		/// The payment hash whose preimage our counterparty needs to claim this HTLC.
 		payment_hash: PaymentHash,
+		/// The payment ID for outbound HTLCs, enabling grouping of MPPs.
+		/// `None` for forwarded HTLCs or if the channel monitor was created before this feature.
+		payment_id: Option<PaymentId>,
 		/// Whether this HTLC represents a payment which was sent outbound from us. Otherwise it
 		/// represents an HTLC which was forwarded (and should, thus, have a corresponding inbound
 		/// edge on another channel).
@@ -999,6 +1002,31 @@ impl Balance {
 			Balance::MaybePreimageClaimableHTLC { .. } => 0,
 		}
 	}
+
+	/// Aggregates outbound payment HTLCs from a list of balances by [`PaymentId`].
+	///
+	/// Returns a map of [`PaymentId`] to total satoshis locked across all channels. For balance
+	/// calculations, prefer [`reconcile_inflight_payments`] which provides an atomic view and
+	/// returns msats.
+	///
+	/// [`reconcile_inflight_payments`]: crate::ln::channelmanager::reconcile_inflight_payments
+	pub fn aggregate_outbound_htlcs_by_payment(balances: &[Balance]) -> HashMap<PaymentId, u64> {
+		let mut payment_amounts = HashMap::default();
+
+		for balance in balances {
+			if let Balance::MaybeTimeoutClaimableHTLC {
+				amount_satoshis,
+				payment_id: Some(payment_id),
+				outbound_payment: true,
+				..
+			} = balance
+			{
+				*payment_amounts.entry(*payment_id).or_insert(0) += amount_satoshis;
+			}
+		}
+
+		payment_amounts
+	}
 }
 
 /// An HTLC which has been irrevocably resolved on-chain, and has reached ANTI_REORG_DELAY.
@@ -2865,15 +2893,16 @@ impl<Signer: EcdsaChannelSigner> ChannelMonitorImpl<Signer> {
 					source: BalanceSource::Htlc,
 				});
 			} else {
-				let outbound_payment = match source {
+				let (outbound_payment, payment_id) = match source {
 					None => panic!("Outbound HTLCs should have a source"),
-					Some(&HTLCSource::PreviousHopData(_)) => false,
-					Some(&HTLCSource::OutboundRoute { .. }) => true,
+					Some(&HTLCSource::PreviousHopData(_)) => (false, None),
+					Some(&HTLCSource::OutboundRoute { payment_id, .. }) => (true, Some(payment_id)),
 				};
 				return Some(Balance::MaybeTimeoutClaimableHTLC {
 					amount_satoshis: htlc.amount_msat / 1000,
 					claimable_height: htlc.cltv_expiry,
 					payment_hash: htlc.payment_hash,
+					payment_id,
 					outbound_payment,
 				});
 			}
@@ -3077,10 +3106,10 @@ impl<Signer: EcdsaChannelSigner> ChannelMonitor<Signer> {
 					htlc.amount_msat
 				} else { htlc.amount_msat % 1000 };
 				if htlc.offered {
-					let outbound_payment = match source {
+					let (outbound_payment, payment_id) = match source {
 						None => panic!("Outbound HTLCs should have a source"),
-						Some(HTLCSource::PreviousHopData(_)) => false,
-						Some(HTLCSource::OutboundRoute { .. }) => true,
+						Some(HTLCSource::PreviousHopData(_)) => (false, None),
+						Some(HTLCSource::OutboundRoute { payment_id, .. }) => (true, Some(*payment_id)),
 					};
 					if outbound_payment {
 						outbound_payment_htlc_rounded_msat += rounded_value_msat;
@@ -3092,6 +3121,7 @@ impl<Signer: EcdsaChannelSigner> ChannelMonitor<Signer> {
 							amount_satoshis: htlc.amount_msat / 1000,
 							claimable_height: htlc.cltv_expiry,
 							payment_hash: htlc.payment_hash,
+							payment_id,
 							outbound_payment,
 						});
 					}

lightning/src/chain/mod.rs

@@ -35,6 +35,9 @@ pub(crate) mod onchaintx;
 pub(crate) mod package;
 pub mod transaction;
 
+#[cfg(test)]
+mod balance_aggregation_tests;
+
 /// The best known block as identified by its hash and height.
 #[derive(Clone, Copy, Debug, Hash, PartialEq, Eq)]
 pub struct BestBlock {