Block RAA `ChannelMonitorUpdate`s on `PaymentClaimed` events

diff --git a/lightning/src/ln/async_signer_tests.rs b/lightning/src/ln/async_signer_tests.rs
index 225f588..511bdce 100644
--- a/lightning/src/ln/async_signer_tests.rs
+++ b/lightning/src/ln/async_signer_tests.rs
@@ -1395,6 +1395,7 @@ fn test_no_disconnect_while_async_commitment_signed_expecting_remote_revoke_and_
 	let (preimage, payment_hash, ..) = route_payment(&nodes[0], &[&nodes[1]], payment_amount);
 	nodes[1].node.claim_funds(preimage);
 	check_added_monitors(&nodes[1], 1);
+	expect_payment_claimed!(nodes[1], payment_hash, payment_amount);
 
 	// We'll disable signing counterparty commitments on the payment sender.
 	nodes[0].disable_channel_signer_op(&node_b_id, &chan_id, SignerOp::SignCounterpartyCommitment);
@@ -1403,6 +1404,7 @@ fn test_no_disconnect_while_async_commitment_signed_expecting_remote_revoke_and_
 	// the `commitment_signed` is no longer pending.
 	let mut update = get_htlc_update_msgs!(&nodes[1], node_a_id);
 	nodes[0].node.handle_update_fulfill_htlc(node_b_id, update.update_fulfill_htlcs.remove(0));
+	expect_payment_sent(&nodes[0], preimage, None, false, false);
 	nodes[0].node.handle_commitment_signed_batch_test(node_b_id, &update.commitment_signed);
 	check_added_monitors(&nodes[0], 1);
 
@@ -1426,7 +1428,4 @@ fn test_no_disconnect_while_async_commitment_signed_expecting_remote_revoke_and_
 	};
 	assert!(nodes[0].node.get_and_clear_pending_msg_events().is_empty());
 	assert!(nodes[1].node.get_and_clear_pending_msg_events().into_iter().any(has_disconnect_event));
-
-	expect_payment_sent(&nodes[0], preimage, None, false, false);
-	expect_payment_claimed!(nodes[1], payment_hash, payment_amount);
 }
diff --git a/lightning/src/ln/chanmon_update_fail_tests.rs b/lightning/src/ln/chanmon_update_fail_tests.rs
index 0a8f258..1302932 100644
--- a/lightning/src/ln/chanmon_update_fail_tests.rs
+++ b/lightning/src/ln/chanmon_update_fail_tests.rs
@@ -4694,6 +4694,23 @@ fn test_single_channel_multiple_mpp() {
 	// `update_fulfill_htlc`/`commitment_signed` pair to pass to our counterparty.
 	do_a_write.send(()).unwrap();
 
+	let event_node: &'static TestChannelManager<'static, 'static> =
+		unsafe { std::mem::transmute(nodes[8].node as &TestChannelManager) };
+	let thrd_event = std::thread::spawn(move || {
+		let mut have_event = false;
+		while !have_event {
+			let mut events = event_node.get_and_clear_pending_events();
+			assert!(events.len() == 1 || events.len() == 0);
+			if events.len() == 1 {
+				if let Event::PaymentClaimed { .. } = events[0] {
+				} else {
+					panic!("Unexpected event {events:?}");
+				}
+				have_event = true;
+			}
+		}
+	});
+
 	// 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
@@ -4713,7 +4730,11 @@ fn test_single_channel_multiple_mpp() {
 	});
 	block_thrd2.store(false, Ordering::Release);
 	let mut first_updates = get_htlc_update_msgs(&nodes[8], &node_h_id);
+
+	// Thread 2 could unblock first, or it could get blocked waiting on us to process a
+	// `PaymentClaimed` event. Either way, wait until both have finished.
 	thrd2.join().unwrap();
+	thrd_event.join().unwrap();
 
 	// Disconnect node 6 from all its peers so it doesn't bother to fail the HTLCs back
 	nodes[7].node.peer_disconnected(node_b_id);
@@ -4760,8 +4781,6 @@ fn test_single_channel_multiple_mpp() {
 	thrd4.join().unwrap();
 	thrd.join().unwrap();
 
-	expect_payment_claimed!(nodes[8], payment_hash, 50_000_000);
-
 	// At the end, we should have 7 ChannelMonitorUpdates - 6 for HTLC claims, and one for the
 	// above `revoke_and_ack`.
 	check_added_monitors(&nodes[8], 7);
diff --git a/lightning/src/ln/channelmanager.rs b/lightning/src/ln/channelmanager.rs
index 55521a8..1fd99f8 100644
--- a/lightning/src/ln/channelmanager.rs
+++ b/lightning/src/ln/channelmanager.rs
@@ -934,9 +934,19 @@ struct ClaimingPayment {
 	sender_intended_value: Option<u64>,
 	onion_fields: Option<RecipientOnionFields>,
 	payment_id: Option<PaymentId>,
+	/// When we claim and generate a [`Event::PaymentClaimed`], we want to block any
+	/// payment-preimage-removing RAA [`ChannelMonitorUpdate`]s until the [`Event::PaymentClaimed`]
+	/// is handled, ensuring we can regenerate the event on restart. We pick a random channel to
+	/// block and store it here.
+	///
+	/// Note that once we disallow downgrades to 0.1 we should be able to simply use
+	/// [`Self::htlcs`] to generate this rather than storing it here (as we won't need the funding
+	/// outpoint), allowing us to remove this field.
+	durable_preimage_channel: Option<(OutPoint, PublicKey, ChannelId)>,
 }
 impl_writeable_tlv_based!(ClaimingPayment, {
 	(0, amount_msat, required),
+	(1, durable_preimage_channel, option),
 	(2, payment_purpose, required),
 	(4, receiver_node_id, required),
 	(5, htlcs, optional_vec),
@@ -1083,6 +1093,16 @@ impl ClaimablePayments {
 					.or_insert_with(|| {
 						let htlcs = payment.htlcs.iter().map(events::ClaimedHTLC::from).collect();
 						let sender_intended_value = payment.htlcs.first().map(|htlc| htlc.total_msat);
+						// Pick an "arbitrary" channel to block RAAs on until the `PaymentSent`
+						// event is processed, specifically the last channel to get claimed.
+						let durable_preimage_channel = payment.htlcs.last().map_or(None, |htlc| {
+							if let Some(node_id) = htlc.prev_hop.counterparty_node_id {
+								Some((htlc.prev_hop.outpoint, node_id, htlc.prev_hop.channel_id))
+							} else {
+								None
+							}
+						});
+						debug_assert!(durable_preimage_channel.is_some());
 						ClaimingPayment {
 							amount_msat: payment.htlcs.iter().map(|source| source.value).sum(),
 							payment_purpose: payment.purpose,
@@ -1091,6 +1111,7 @@ impl ClaimablePayments {
 							sender_intended_value,
 							onion_fields: payment.onion_fields,
 							payment_id: Some(payment_id),
+							durable_preimage_channel,
 						}
 					}).clone();
 
@@ -8704,6 +8725,7 @@ This indicates a bug inside LDK. Please report this error at https://github.com/
 						sender_intended_value: sender_intended_total_msat,
 						onion_fields,
 						payment_id,
+						durable_preimage_channel,
 					}) = payment {
 						let event = events::Event::PaymentClaimed {
 							payment_hash,
@@ -8715,7 +8737,18 @@ This indicates a bug inside LDK. Please report this error at https://github.com/
 							onion_fields,
 							payment_id,
 						};
-						let event_action = (event, None);
+						let action = if let Some((outpoint, counterparty_node_id, channel_id))
+							= durable_preimage_channel
+						{
+							Some(EventCompletionAction::ReleaseRAAChannelMonitorUpdate {
+								channel_funding_outpoint: Some(outpoint),
+								counterparty_node_id,
+								channel_id,
+							})
+						} else {
+							None
+						};
+						let event_action = (event, action);
 						let mut pending_events = self.pending_events.lock().unwrap();
 						// If we're replaying a claim on startup we may end up duplicating an event
 						// that's already in our queue, so check before we push another one. The
@@ -17104,6 +17137,10 @@ where
 								onion_fields: payment.onion_fields,
 								payment_id: Some(payment_id),
 							},
+							// Note that we don't bother adding a EventCompletionAction here to
+							// ensure the `PaymentClaimed` event is durable processed as this
+							// should only be hit for particularly old channels and we don't have
+							// enough information to generate such an action.
 							None,
 						));
 					}
diff --git a/lightning/src/ln/monitor_tests.rs b/lightning/src/ln/monitor_tests.rs
index 13b9301..c903424 100644
--- a/lightning/src/ln/monitor_tests.rs
+++ b/lightning/src/ln/monitor_tests.rs
@@ -3211,3 +3211,65 @@ fn test_update_replay_panics() {
 	monitor.update_monitor(&updates[2], &nodes[1].tx_broadcaster, &nodes[1].fee_estimator, &nodes[1].logger).unwrap();
 	monitor.update_monitor(&updates[3], &nodes[1].tx_broadcaster, &nodes[1].fee_estimator, &nodes[1].logger).unwrap();
 }
+
+#[test]
+fn test_claim_event_never_handled() {
+	// When a payment is claimed, the `ChannelMonitorUpdate` containing the payment preimage goes
+	// out and when it completes the `PaymentClaimed` event is generated. If the channel then
+	// progresses forward a few steps, the payment preimage will then eventually be removed from
+	// the channel. By that point, we have to make sure that the `PaymentClaimed` event has been
+	// handled (which ensures the user has maked the payment received).
+	// Otherwise, it is possible that, on restart, we load with a stale `ChannelManager` which
+	// doesn't have the `PaymentClaimed` event and it needs to rebuild it from the
+	// `ChannelMonitor`'s payment information and preimage.
+	let chanmon_cfgs = create_chanmon_cfgs(2);
+	let node_cfgs = create_node_cfgs(2, &chanmon_cfgs);
+	let persister;
+	let new_chain_mon;
+	let node_chanmgrs = create_node_chanmgrs(2, &node_cfgs, &[None, None]);
+	let nodes_1_reload;
+	let mut nodes = create_network(2, &node_cfgs, &node_chanmgrs);
+
+	let node_a_id = nodes[0].node.get_our_node_id();
+	let node_b_id = nodes[1].node.get_our_node_id();
+
+	let init_node_ser = nodes[1].node.encode();
+
+	let chan = create_announced_chan_between_nodes(&nodes, 0, 1);
+
+	// Send the payment we'll ultimately test the PaymentClaimed event for.
+	let (preimage_a, payment_hash_a, ..) = route_payment(&nodes[0], &[&nodes[1]], 1_000_000);
+
+	nodes[1].node.claim_funds(preimage_a);
+	check_added_monitors(&nodes[1], 1);
+
+	let mut updates = get_htlc_update_msgs(&nodes[1], &node_a_id);
+	nodes[0].node.handle_update_fulfill_htlc(node_b_id, updates.update_fulfill_htlcs.remove(0));
+	expect_payment_sent(&nodes[0], preimage_a, None, false, false);
+
+	nodes[0].node.handle_commitment_signed_batch_test(node_b_id, &updates.commitment_signed);
+	check_added_monitors(&nodes[0], 1);
+
+	// Once the `PaymentClaimed` event is generated, further RAA `ChannelMonitorUpdate`s will be
+	// blocked until it is handled, ensuring we never get far enough to remove the preimage.
+	let (raa, cs) = get_revoke_commit_msgs(&nodes[0], &node_b_id);
+	nodes[1].node.handle_revoke_and_ack(node_a_id, &raa);
+	nodes[1].node.handle_commitment_signed_batch_test(node_a_id, &cs);
+	check_added_monitors(&nodes[1], 0);
+
+	// The last RAA here should be blocked waiting on us to handle the PaymentClaimed event before
+	// continuing. Otherwise, we'd be able to make enough progress that the payment preimage is
+	// removed from node A's `ChannelMonitor`. This leaves us unable to make further progress.
+	assert!(nodes[1].node.get_and_clear_pending_msg_events().is_empty());
+
+	// Finally, reload node B with an empty `ChannelManager` and check that we get the
+	// `PaymentClaimed` event.
+	let chan_0_monitor_serialized = get_monitor!(nodes[1], chan.2).encode();
+	let mons = &[&chan_0_monitor_serialized[..]];
+	reload_node!(nodes[1], &init_node_ser, mons, persister, new_chain_mon, nodes_1_reload);
+
+	expect_payment_claimed!(nodes[1], payment_hash_a, 1_000_000);
+	// The reload logic spuriously generates a redundant payment preimage-containing
+	// `ChannelMonitorUpdate`.
+	check_added_monitors(&nodes[1], 2);
+}
diff --git a/lightning/src/ln/quiescence_tests.rs b/lightning/src/ln/quiescence_tests.rs
index 17b6535..211e79a 100644
--- a/lightning/src/ln/quiescence_tests.rs
+++ b/lightning/src/ln/quiescence_tests.rs
@@ -197,6 +197,7 @@ fn test_quiescence_waits_for_async_signer_and_monitor_update() {
 	let (preimage, payment_hash, ..) = route_payment(&nodes[0], &[&nodes[1]], payment_amount);
 	nodes[1].node.claim_funds(preimage);
 	check_added_monitors(&nodes[1], 1);
+	expect_payment_claimed!(&nodes[1], payment_hash, payment_amount);
 
 	let mut update = get_htlc_update_msgs!(&nodes[1], node_id_0);
 	nodes[0].node.handle_update_fulfill_htlc(node_id_1, update.update_fulfill_htlcs.remove(0));
@@ -223,8 +224,6 @@ fn test_quiescence_waits_for_async_signer_and_monitor_update() {
 	nodes[1].enable_channel_signer_op(&node_id_0, &chan_id, SignerOp::ReleaseCommitmentSecret);
 	nodes[1].node.signer_unblocked(Some((node_id_0, chan_id)));
 
-	expect_payment_claimed!(&nodes[1], payment_hash, payment_amount);
-
 	macro_rules! find_msg {
 		($events: expr, $msg: ident) => {{
 			$events
@@ -418,14 +417,11 @@ fn quiescence_updates_go_to_holding_cell(fail_htlc: bool) {
 	if fail_htlc {
 		nodes[0].node.handle_update_fail_htlc(node_id_1, &update.update_fail_htlcs[0]);
 	} else {
+		expect_payment_claimed!(nodes[1], payment_hash2, payment_amount);
 		nodes[0].node.handle_update_fulfill_htlc(node_id_1, update.update_fulfill_htlcs.remove(0));
 	}
 	commitment_signed_dance!(&nodes[0], &nodes[1], update.commitment_signed, false);
 
-	if !fail_htlc {
-		expect_payment_claimed!(nodes[1], payment_hash2, payment_amount);
-	}
-
 	// The payment from nodes[0] should now be seen as failed/successful.
 	let events = nodes[0].node.get_and_clear_pending_events();
 	assert_eq!(events.len(), 2);
@@ -454,6 +450,7 @@ fn quiescence_updates_go_to_holding_cell(fail_htlc: bool) {
 	if fail_htlc {
 		nodes[1].node.handle_update_fail_htlc(node_id_0, &update.update_fail_htlcs[0]);
 	} else {
+		expect_payment_claimed!(nodes[0], payment_hash1, payment_amount);
 		nodes[1].node.handle_update_fulfill_htlc(node_id_0, update.update_fulfill_htlcs.remove(0));
 	}
 	commitment_signed_dance!(&nodes[1], &nodes[0], update.commitment_signed, false);
@@ -463,7 +460,6 @@ fn quiescence_updates_go_to_holding_cell(fail_htlc: bool) {
 		let conditions = PaymentFailedConditions::new();
 		expect_payment_failed_conditions(&nodes[1], payment_hash1, true, conditions);
 	} else {
-		expect_payment_claimed!(nodes[0], payment_hash1, payment_amount);
 		expect_payment_sent(&nodes[1], payment_preimage1, None, true, true);
 	}
 }