Do not track HTLC IDs as separate MPP parts which need claiming

TheBlueMatt · TheBlueMatt · commit 47c147431723 · 2025-03-22T00:44:47.000Z
When we claim an MPP payment, we need to track which channels have
had the preimage durably added to their `ChannelMonitor` to ensure
we don't remove the preimage from any `ChannelMonitor`s until all
`ChannelMonitor`s have the preimage.

Previously, we tracked each MPP part, down to the HTLC ID, as a
part which we needed to get the preimage on disk for. However, this
is not necessary - once a `ChannelMonitor` has a preimage, it
applies it to all inbound HTLCs with the same payment hash.

Further, this can cause a channel to wait on itself in cases of
high-latency synchronous persistence -
 * If we have receive an MPP payment for which multiple parts came
   to us over the same channel,
 * and claim the MPP payment, creating a `ChannelMonitorUpdate` for
   the first part but enqueueing the remaining HTLC claim(s) in the
   channel's holding cell,
 * and we receive a `revoke_and_ack` for the same channel before
   the `ChannelManager::claim_payment` method completes (as each
   claim waits for the `ChannelMonitorUpdate` persistence),
 * we will cause the `ChannelMonitorUpdate` for that
   `revoke_and_ack` to go into the blocked set, waiting on the MPP
   parts to be fully claimed,
 * but when `claim_payment` goes to add the next
   `ChannelMonitorUpdate` for the MPP claim, it will be placed in
   the blocked set, since the blocked set is non-empty.

Thus, we'll end up with a `ChannelMonitorUpdate` in the blocked set
which is needed to unblock the channel since it is a part of the
MPP set which blocked the channel.
diff --git a/lightning/src/ln/chanmon_update_fail_tests.rs b/lightning/src/ln/chanmon_update_fail_tests.rs
@@ -3860,3 +3860,145 @@ fn test_claim_to_closed_channel_blocks_claimed_event() {
 	nodes[1].chain_monitor.complete_sole_pending_chan_update(&chan_a.2);
 	expect_payment_claimed!(nodes[1], payment_hash, 1_000_000);
 }
+
+#[test]
+#[cfg(feature = "std")]
+fn test_single_channel_multiple_mpp() {
+	// Test what happens when we attempt to claim an MPP with many parts that came to us through
+	// the same channel with a synchronous persistence interface which has very high latency.
+	//
+	// Previously, if a `revoke_and_ack` came in while we were still running in
+	// `ChannelManager::claim_payment` we'd end up hanging waiting to apply a
+	// `ChannelMonitorUpdate` until after it completed. See the commit which introduced this test
+	// for more info.
+	let chanmon_cfgs = create_chanmon_cfgs(7);
+	let node_cfgs = create_node_cfgs(7, &chanmon_cfgs);
+	let node_chanmgrs = create_node_chanmgrs(7, &node_cfgs, &[None; 7]);
+	let mut nodes = create_network(7, &node_cfgs, &node_chanmgrs);
+
+	let node_5_id = nodes[5].node.get_our_node_id();
+	let node_6_id = nodes[6].node.get_our_node_id();
+
+	// Send an MPP payment in four parts along the path shown from top to bottom
+	//      0
+	//   1 2 3 4
+	//      5
+	//      6
+
+	create_announced_chan_between_nodes_with_value(&nodes, 0, 1, 100_000, 0);
+	create_announced_chan_between_nodes_with_value(&nodes, 0, 2, 100_000, 0);
+	create_announced_chan_between_nodes_with_value(&nodes, 0, 3, 100_000, 0);
+	create_announced_chan_between_nodes_with_value(&nodes, 0, 4, 100_000, 0);
+	create_announced_chan_between_nodes_with_value(&nodes, 1, 5, 100_000, 0);
+	create_announced_chan_between_nodes_with_value(&nodes, 2, 5, 100_000, 0);
+	create_announced_chan_between_nodes_with_value(&nodes, 3, 5, 100_000, 0);
+	create_announced_chan_between_nodes_with_value(&nodes, 4, 5, 100_000, 0);
+	create_announced_chan_between_nodes_with_value(&nodes, 5, 6, 1_000_000, 0);
+
+	let (mut route, payment_hash, payment_preimage, payment_secret) = get_route_and_payment_hash!(&nodes[0], nodes[6], 30_000_000);
+
+	send_along_route_with_secret(&nodes[0], route, &[&[&nodes[1], &nodes[5], &nodes[6]], &[&nodes[2], &nodes[5], &nodes[6]], &[&nodes[3], &nodes[5], &nodes[6]], &[&nodes[4], &nodes[5], &nodes[6]]], 30_000_000, payment_hash, payment_secret);
+
+	let (do_a_write, blocker) = std::sync::mpsc::sync_channel(0);
+	*nodes[6].chain_monitor.write_blocker.lock().unwrap() = Some(blocker);
+
+	// Until we have std::thread::scoped we have to unsafe { turn off the borrow checker }.
+	let claim_node: &'static TestChannelManager<'static, 'static> =
+		unsafe { std::mem::transmute(nodes[6].node as &TestChannelManager) };
+	let thrd = std::thread::spawn(move || {
+		// Initiate the claim in a background thread as it will immediately block waiting on the
+		// `write_blocker` we set above.
+		claim_node.claim_funds(payment_preimage);
+	});
+
+	// First unlock one monitor so that we have a pending
+	// `update_fulfill_htlc`/`commitment_signed` pair to pass to our counterparty.
+	do_a_write.send(()).unwrap();
+
+	// Then fetch the `update_fulfill_htlc`/`commitment_signed`. Note that the
+	// `get_and_clear_pending_msg_events` will immediately hang trying to take a peer lock which
+	// `claim_funds` is holding. Thus, we release a second write after a small sleep in the
+	// background to give `claim_funds` a chance to step forward, unblocking
+	// `get_and_clear_pending_msg_events`.
+	const MAX_THREAD_INIT_TIME: std::time::Duration = std::time::Duration::from_millis(10);
+	let do_a_write_background = do_a_write.clone();
+	let thrd2 = std::thread::spawn(move || {
+		std::thread::sleep(MAX_THREAD_INIT_TIME);
+		do_a_write_background.send(()).unwrap();
+	});
+	let first_updates = get_htlc_update_msgs(&nodes[6], &nodes[5].node.get_our_node_id());
+	thrd2.join().unwrap();
+
+	nodes[5].node.peer_disconnected(nodes[1].node.get_our_node_id());
+	nodes[5].node.peer_disconnected(nodes[2].node.get_our_node_id());
+	nodes[5].node.peer_disconnected(nodes[3].node.get_our_node_id());
+	nodes[5].node.peer_disconnected(nodes[4].node.get_our_node_id());
+
+	nodes[5].node.handle_update_fulfill_htlc(node_6_id, &first_updates.update_fulfill_htlcs[0]);
+	check_added_monitors(&nodes[5], 1);
+	expect_payment_forwarded!(nodes[5], nodes[1], nodes[6], Some(1000), false, false);
+	nodes[5].node.handle_commitment_signed(node_6_id, &first_updates.commitment_signed);
+	check_added_monitors(&nodes[5], 1);
+	let (raa, cs) = get_revoke_commit_msgs(&nodes[5], &node_6_id);
+
+	// Now, handle the `revoke_and_ack` from node 5. Note that `claim_funds` is still blocked on
+	// our peer lock, so we have to release a write to let it process.
+	let do_a_write_background = do_a_write.clone();
+	let thrd3 = std::thread::spawn(move || {
+		std::thread::sleep(MAX_THREAD_INIT_TIME);
+		do_a_write_background.send(()).unwrap();
+	});
+	nodes[6].node.handle_revoke_and_ack(node_5_id, &raa);
+	thrd3.join().unwrap();
+
+	let thrd4 = std::thread::spawn(move || {
+		std::thread::sleep(MAX_THREAD_INIT_TIME);
+		do_a_write.send(()).unwrap();
+		do_a_write.send(()).unwrap();
+	});
+
+	thrd4.join().unwrap();
+	thrd.join().unwrap();
+
+	expect_payment_claimed!(nodes[6], payment_hash, 30_000_000);
+
+	// At the end, we should have 5 ChannelMonitorUpdates - 4 for HTLC claims, and one for the
+	// above `revoke_and_ack`.
+	check_added_monitors(&nodes[6], 5);
+
+	// Now drive everything to the end, at least as far as node 6 is concerned...
+	*nodes[6].chain_monitor.write_blocker.lock().unwrap() = None;
+	nodes[6].node.handle_commitment_signed(node_5_id, &cs);
+	check_added_monitors(&nodes[6], 1);
+
+	let (updates, raa) = get_updates_and_revoke(&nodes[6], &nodes[5].node.get_our_node_id());
+	nodes[5].node.handle_update_fulfill_htlc(node_6_id, &updates.update_fulfill_htlcs[0]);
+	expect_payment_forwarded!(nodes[5], nodes[2], nodes[6], Some(1000), false, false);
+	nodes[5].node.handle_update_fulfill_htlc(node_6_id, &updates.update_fulfill_htlcs[1]);
+	expect_payment_forwarded!(nodes[5], nodes[3], nodes[6], Some(1000), false, false);
+	nodes[5].node.handle_commitment_signed(node_6_id, &updates.commitment_signed);
+	nodes[5].node.handle_revoke_and_ack(node_6_id, &raa);
+	check_added_monitors(&nodes[5], 4);
+
+	let (raa, cs) = get_revoke_commit_msgs(&nodes[5], &node_6_id);
+
+	nodes[6].node.handle_revoke_and_ack(node_5_id, &raa);
+	nodes[6].node.handle_commitment_signed(node_5_id, &cs);
+	check_added_monitors(&nodes[6], 2);
+
+	let (updates, raa) = get_updates_and_revoke(&nodes[6], &nodes[5].node.get_our_node_id());
+	nodes[5].node.handle_update_fulfill_htlc(node_6_id, &updates.update_fulfill_htlcs[0]);
+	expect_payment_forwarded!(nodes[5], nodes[4], nodes[6], Some(1000), false, false);
+	nodes[5].node.handle_commitment_signed(node_6_id, &updates.commitment_signed);
+	nodes[5].node.handle_revoke_and_ack(node_6_id, &raa);
+	check_added_monitors(&nodes[5], 3);
+
+	let (raa, cs) = get_revoke_commit_msgs(&nodes[5], &node_6_id);
+	nodes[6].node.handle_revoke_and_ack(node_5_id, &raa);
+	nodes[6].node.handle_commitment_signed(node_5_id, &cs);
+	check_added_monitors(&nodes[6], 2);
+
+	let raa = get_event_msg!(nodes[6], MessageSendEvent::SendRevokeAndACK, node_5_id);
+	nodes[5].node.handle_revoke_and_ack(node_6_id, &raa);
+	check_added_monitors(&nodes[5], 1);
+}
diff --git a/lightning/src/ln/channelmanager.rs b/lightning/src/ln/channelmanager.rs
@@ -1132,7 +1132,7 @@ pub(crate) enum MonitorUpdateCompletionAction {
 		/// A pending MPP claim which hasn't yet completed.
 		///
 		/// Not written to disk.
-		pending_mpp_claim: Option<(PublicKey, ChannelId, u64, PendingMPPClaimPointer)>,
+		pending_mpp_claim: Option<(PublicKey, ChannelId, PendingMPPClaimPointer)>,
 	},
 	/// Indicates an [`events::Event`] should be surfaced to the user and possibly resume the
 	/// operation of another channel.
@@ -1234,10 +1234,16 @@ impl From<&MPPClaimHTLCSource> for HTLCClaimSource {
 	}
 }
 
+#[derive(Debug)]
+pub(crate) struct PendingMPPClaim {
+	channels_without_preimage: Vec<(PublicKey, OutPoint, ChannelId)>,
+	channels_with_preimage: Vec<(PublicKey, OutPoint, ChannelId)>,
+}
+
 #[derive(Clone, Debug, Hash, PartialEq, Eq)]
 /// The source of an HTLC which is being claimed as a part of an incoming payment. Each part is
-/// tracked in [`PendingMPPClaim`] as well as in [`ChannelMonitor`]s, so that it can be converted
-/// to an [`HTLCClaimSource`] for claim replays on startup.
+/// tracked in [`ChannelMonitor`]s, so that it can be converted to an [`HTLCClaimSource`] for claim
+/// replays on startup.
 struct MPPClaimHTLCSource {
 	counterparty_node_id: PublicKey,
 	funding_txo: OutPoint,
@@ -1252,12 +1258,6 @@ impl_writeable_tlv_based!(MPPClaimHTLCSource, {
 	(6, htlc_id, required),
 });
 
-#[derive(Debug)]
-pub(crate) struct PendingMPPClaim {
-	channels_without_preimage: Vec<MPPClaimHTLCSource>,
-	channels_with_preimage: Vec<MPPClaimHTLCSource>,
-}
-
 #[derive(Clone, Debug, PartialEq, Eq)]
 /// When we're claiming a(n MPP) payment, we want to store information about that payment in the
 /// [`ChannelMonitor`] so that we can replay the claim without any information from the
@@ -7207,8 +7207,17 @@ where
 				}
 			}).collect();
 			let pending_mpp_claim_ptr_opt = if sources.len() > 1 {
+				let mut channels_without_preimage = Vec::with_capacity(mpp_parts.len());
+				for source in sources.iter() {
+					if let Some(cp_id) = source.prev_hop.counterparty_node_id {
+						let chan = (cp_id, source.prev_hop.outpoint, source.prev_hop.channel_id);
+						if !channels_without_preimage.contains(&chan) {
+							channels_without_preimage.push(chan);
+						}
+					}
+				}
 				Some(Arc::new(Mutex::new(PendingMPPClaim {
-					channels_without_preimage: mpp_parts.clone(),
+					channels_without_preimage,
 					channels_with_preimage: Vec::new(),
 				})))
 			} else {
@@ -7219,7 +7228,7 @@ where
 				let this_mpp_claim = pending_mpp_claim_ptr_opt.as_ref().and_then(|pending_mpp_claim|
 					if let Some(cp_id) = htlc.prev_hop.counterparty_node_id {
 						let claim_ptr = PendingMPPClaimPointer(Arc::clone(pending_mpp_claim));
-						Some((cp_id, htlc.prev_hop.channel_id, htlc.prev_hop.htlc_id, claim_ptr))
+						Some((cp_id, htlc.prev_hop.channel_id, claim_ptr))
 					} else {
 						None
 					}
@@ -7552,7 +7561,7 @@ This indicates a bug inside LDK. Please report this error at https://github.com/
 		for action in actions.into_iter() {
 			match action {
 				MonitorUpdateCompletionAction::PaymentClaimed { payment_hash, pending_mpp_claim } => {
-					if let Some((counterparty_node_id, chan_id, htlc_id, claim_ptr)) = pending_mpp_claim {
+					if let Some((counterparty_node_id, chan_id, claim_ptr)) = pending_mpp_claim {
 						let per_peer_state = self.per_peer_state.read().unwrap();
 						per_peer_state.get(&counterparty_node_id).map(|peer_state_mutex| {
 							let mut peer_state = peer_state_mutex.lock().unwrap();
@@ -7563,24 +7572,17 @@ This indicates a bug inside LDK. Please report this error at https://github.com/
 										if *pending_claim == claim_ptr {
 											let mut pending_claim_state_lock = pending_claim.0.lock().unwrap();
 											let pending_claim_state = &mut *pending_claim_state_lock;
-											pending_claim_state.channels_without_preimage.retain(|htlc_info| {
+											pending_claim_state.channels_without_preimage.retain(|(cp, op, cid)| {
 												let this_claim =
-													htlc_info.counterparty_node_id == counterparty_node_id
-													&& htlc_info.channel_id == chan_id
-													&& htlc_info.htlc_id == htlc_id;
+													*cp == counterparty_node_id && *cid == chan_id;
 												if this_claim {
-													pending_claim_state.channels_with_preimage.push(htlc_info.clone());
+													pending_claim_state.channels_with_preimage.push((*cp, *op, *cid));
 													false
 												} else { true }
 											});
 											if pending_claim_state.channels_without_preimage.is_empty() {
-												for htlc_info in pending_claim_state.channels_with_preimage.iter() {
-													let freed_chan = (
-														htlc_info.counterparty_node_id,
-														htlc_info.funding_txo,
-														htlc_info.channel_id,
-														blocker.clone()
-													);
+												for (cp, op, cid) in pending_claim_state.channels_with_preimage.iter() {
+													let freed_chan = (*cp, *op, *cid, blocker.clone());
 													freed_channels.push(freed_chan);
 												}
 											}
@@ -14786,8 +14788,13 @@ where
 						if payment_claim.mpp_parts.is_empty() {
 							return Err(DecodeError::InvalidValue);
 						}
+						let mut channels_without_preimage = payment_claim.mpp_parts.iter()
+							.map(|htlc_info| (htlc_info.counterparty_node_id, htlc_info.funding_txo, htlc_info.channel_id))
+							.collect::<Vec<_>>();
+						channels_without_preimage.sort_unstable();
+						channels_without_preimage.dedup();
 						let pending_claims = PendingMPPClaim {
-							channels_without_preimage: payment_claim.mpp_parts.clone(),
+							channels_without_preimage,
 							channels_with_preimage: Vec::new(),
 						};
 						let pending_claim_ptr_opt = Some(Arc::new(Mutex::new(pending_claims)));
@@ -14820,7 +14827,7 @@ where
 
 						for part in payment_claim.mpp_parts.iter() {
 							let pending_mpp_claim = pending_claim_ptr_opt.as_ref().map(|ptr| (
-								part.counterparty_node_id, part.channel_id, part.htlc_id,
+								part.counterparty_node_id, part.channel_id,
 								PendingMPPClaimPointer(Arc::clone(&ptr))
 							));
 							let pending_claim_ptr = pending_claim_ptr_opt.as_ref().map(|ptr|
diff --git a/lightning/src/ln/functional_test_utils.rs b/lightning/src/ln/functional_test_utils.rs
@@ -779,6 +779,26 @@ pub fn get_revoke_commit_msgs<CM: AChannelManager, H: NodeHolder<CM=CM>>(node: &
 	})
 }
 
+/// Gets a `revoke_and_ack` and `UpdateHTLCs` (i.e. after we get a responding `commitment_signed`
+/// while we have updates in the holding cell).
+pub fn get_updates_and_revoke<CM: AChannelManager, H: NodeHolder<CM=CM>>(node: &H, recipient: &PublicKey) -> (msgs::CommitmentUpdate, msgs::RevokeAndACK) {
+	let events = node.node().get_and_clear_pending_msg_events();
+	assert_eq!(events.len(), 2);
+	(match events[0] {
+		MessageSendEvent::UpdateHTLCs { ref node_id, ref updates } => {
+			assert_eq!(node_id, recipient);
+			(*updates).clone()
+		},
+		_ => panic!("Unexpected event"),
+	}, match events[1] {
+		MessageSendEvent::SendRevokeAndACK { ref node_id, ref msg } => {
+			assert_eq!(node_id, recipient);
+			(*msg).clone()
+		},
+		_ => panic!("Unexpected event"),
+	})
+}
+
 #[macro_export]
 /// Gets an RAA and CS which were sent in response to a commitment update
 ///
diff --git a/lightning/src/util/test_utils.rs b/lightning/src/util/test_utils.rs
@@ -398,12 +398,15 @@ pub struct TestChainMonitor<'a> {
 	/// If this is set to Some(), the next round trip serialization check will not hold after an
 	/// update_channel call (not watch_channel) for the given channel_id.
 	pub expect_monitor_round_trip_fail: Mutex<Option<ChannelId>>,
+	#[cfg(feature = "std")]
+	pub write_blocker: Mutex<Option<std::sync::mpsc::Receiver<()>>>,
 }
 impl<'a> TestChainMonitor<'a> {
 	pub fn new(
 		chain_source: Option<&'a TestChainSource>,
 		broadcaster: &'a (dyn chaininterface::BroadcasterInterface + Sync), logger: &'a TestLogger,
-		fee_estimator: &'a TestFeeEstimator, persister: &'a (dyn Persist<TestChannelSigner> + Sync),
+		fee_estimator: &'a TestFeeEstimator,
+		persister: &'a (dyn Persist<TestChannelSigner> + Sync),
 		keys_manager: &'a TestKeysInterface,
 	) -> Self {
 		Self {
@@ -420,6 +423,8 @@ impl<'a> TestChainMonitor<'a> {
 			keys_manager,
 			expect_channel_force_closed: Mutex::new(None),
 			expect_monitor_round_trip_fail: Mutex::new(None),
+			#[cfg(feature = "std")]
+			write_blocker: Mutex::new(None),
 		}
 	}
 
@@ -433,6 +438,11 @@ impl<'a> chain::Watch<TestChannelSigner> for TestChainMonitor<'a> {
 	fn watch_channel(
 		&self, channel_id: ChannelId, monitor: ChannelMonitor<TestChannelSigner>,
 	) -> Result<chain::ChannelMonitorUpdateStatus, ()> {
+		#[cfg(feature = "std")]
+		if let Some(blocker) = &*self.write_blocker.lock().unwrap() {
+			blocker.recv().unwrap();
+		}
+
 		// At every point where we get a monitor update, we should be able to send a useful monitor
 		// to a watchtower and disk...
 		let mut w = TestVecWriter(Vec::new());
@@ -455,6 +465,11 @@ impl<'a> chain::Watch<TestChannelSigner> for TestChainMonitor<'a> {
 	fn update_channel(
 		&self, channel_id: ChannelId, update: &ChannelMonitorUpdate,
 	) -> chain::ChannelMonitorUpdateStatus {
+		#[cfg(feature = "std")]
+		if let Some(blocker) = &*self.write_blocker.lock().unwrap() {
+			blocker.recv().unwrap();
+		}
+
 		// Every monitor update should survive roundtrip
 		let mut w = TestVecWriter(Vec::new());
 		update.write(&mut w).unwrap();
