Generate new `ReleasePaymentComplete` monitor updates

diff --git a/lightning/src/chain/channelmonitor.rs b/lightning/src/chain/channelmonitor.rs
index e1a0f40..b16bf0c 100644
--- a/lightning/src/chain/channelmonitor.rs
+++ b/lightning/src/chain/channelmonitor.rs
@@ -4119,6 +4119,7 @@ impl<Signer: EcdsaChannelSigner> ChannelMonitorImpl<Signer> {
 		if updates.update_id == LEGACY_CLOSED_CHANNEL_UPDATE_ID || self.lockdown_from_offchain {
 			assert_eq!(updates.updates.len(), 1);
 			match updates.updates[0] {
+				ChannelMonitorUpdateStep::ReleasePaymentComplete { .. } => {},
 				ChannelMonitorUpdateStep::ChannelForceClosed { .. } => {},
 				// We should have already seen a `ChannelForceClosed` update if we're trying to
 				// provide a preimage at this point.
diff --git a/lightning/src/ln/chanmon_update_fail_tests.rs b/lightning/src/ln/chanmon_update_fail_tests.rs
index 4aa587a..e0de92c 100644
--- a/lightning/src/ln/chanmon_update_fail_tests.rs
+++ b/lightning/src/ln/chanmon_update_fail_tests.rs
@@ -3870,6 +3870,7 @@ fn do_test_durable_preimages_on_closed_channel(
 	};
 	nodes[0].node.force_close_broadcasting_latest_txn(&chan_id_ab, &node_b_id, err_msg).unwrap();
 	check_closed_event(&nodes[0], 1, reason, false, &[node_b_id], 100000);
+	check_added_monitors(&nodes[0], 1);
 	let as_closing_tx = nodes[0].tx_broadcaster.txn_broadcasted.lock().unwrap().split_off(0);
 	assert_eq!(as_closing_tx.len(), 1);
 
diff --git a/lightning/src/ln/channelmanager.rs b/lightning/src/ln/channelmanager.rs
index 9c68a0e..e76d207 100644
--- a/lightning/src/ln/channelmanager.rs
+++ b/lightning/src/ln/channelmanager.rs
@@ -1267,6 +1267,12 @@ pub(crate) enum EventCompletionAction {
 		channel_funding_outpoint: Option<OutPoint>,
 		channel_id: ChannelId,
 	},
+
+	/// When a payment's resolution is communicated to the downstream logic via
+	/// [`Event::PaymentSent`] or [`Event::PaymentFailed`] we may want to mark the payment as
+	/// fully-resolved in the [`ChannelMonitor`], which we do via this action.
+	/// Note that this action will be dropped on downgrade to LDK prior to 0.2!
+	ReleasePaymentCompleteChannelMonitorUpdate(PaymentCompleteUpdate),
 }
 impl_writeable_tlv_based_enum!(EventCompletionAction,
 	(0, ReleaseRAAChannelMonitorUpdate) => {
@@ -1279,6 +1285,7 @@ impl_writeable_tlv_based_enum!(EventCompletionAction,
 			ChannelId::v1_from_funding_outpoint(channel_funding_outpoint.unwrap())
 		})),
 	}
+	{1, ReleasePaymentCompleteChannelMonitorUpdate} => (),
 );
 
 /// The source argument which is passed to [`ChannelManager::claim_mpp_part`].
@@ -8009,11 +8016,20 @@ where
 				if let Some(update) = from_monitor_update_completion {
 					// If `fail_htlc` didn't `take` the post-event action, we should go ahead and
 					// complete it here as the failure was duplicative - we've already handled it.
-					// This should mostly only happen on startup, but it is possible to hit it in
-					// rare cases where a MonitorUpdate is replayed after restart because a
-					// ChannelMonitor wasn't persisted after it was applied (even though the
-					// ChannelManager was).
-					// TODO
+					// This can happen in rare cases where a MonitorUpdate is replayed after
+					// restart because a ChannelMonitor wasn't persisted after it was applied (even
+					// though the ChannelManager was).
+					// For such cases, we also check that there's no existing pending event to
+					// complete this action already, which we let finish instead.
+					let action =
+						EventCompletionAction::ReleasePaymentCompleteChannelMonitorUpdate(update);
+					let have_action = {
+						let pending_events = self.pending_events.lock().unwrap();
+						pending_events.iter().any(|(_, act)| act.as_ref() == Some(&action))
+					};
+					if !have_action {
+						self.handle_post_event_actions([action]);
+					}
 				}
 			},
 			HTLCSource::PreviousHopData(HTLCPreviousHopData {
@@ -8642,19 +8658,34 @@ This indicates a bug inside LDK. Please report this error at https://github.com/
 		next_user_channel_id: Option<u128>, attribution_data: Option<AttributionData>,
 		send_timestamp: Option<Duration>,
 	) {
+		debug_assert_eq!(
+			startup_replay,
+			!self.background_events_processed_since_startup.load(Ordering::Acquire)
+		);
+		let htlc_id = SentHTLCId::from_source(&source);
 		match source {
 			HTLCSource::OutboundRoute {
 				session_priv, payment_id, path, bolt12_invoice, ..
 			} => {
-				debug_assert!(self.background_events_processed_since_startup.load(Ordering::Acquire),
+				debug_assert!(!startup_replay,
 					"We don't support claim_htlc claims during startup - monitors may not be available yet");
 				debug_assert_eq!(next_channel_counterparty_node_id, path.hops[0].pubkey);
-				let mut ev_completion_action =
+
+				let mut ev_completion_action = if from_onchain {
+					let release = PaymentCompleteUpdate {
+						counterparty_node_id: next_channel_counterparty_node_id,
+						channel_funding_outpoint: next_channel_outpoint,
+						channel_id: next_channel_id,
+						htlc_id,
+					};
+					Some(EventCompletionAction::ReleasePaymentCompleteChannelMonitorUpdate(release))
+				} else {
 					Some(EventCompletionAction::ReleaseRAAChannelMonitorUpdate {
 						channel_funding_outpoint: Some(next_channel_outpoint),
 						channel_id: next_channel_id,
 						counterparty_node_id: path.hops[0].pubkey,
-					});
+					})
+				};
 				self.pending_outbound_payments.claim_htlc(
 					payment_id,
 					payment_preimage,
@@ -11372,12 +11403,18 @@ This indicates a bug inside LDK. Please report this error at https://github.com/
 								channel_id,
 							};
 							let reason = HTLCFailReason::from_failure_code(failure_reason);
+							let completion_update = Some(PaymentCompleteUpdate {
+								counterparty_node_id,
+								channel_funding_outpoint: funding_outpoint,
+								channel_id,
+								htlc_id: SentHTLCId::from_source(&htlc_update.source),
+							});
 							self.fail_htlc_backwards_internal(
 								&htlc_update.source,
 								&htlc_update.payment_hash,
 								&reason,
 								receiver,
-								None,
+								completion_update,
 							);
 						}
 					},
@@ -12864,8 +12901,62 @@ where
 					channel_id,
 					counterparty_node_id,
 				} => {
+					let startup_complete =
+						self.background_events_processed_since_startup.load(Ordering::Acquire);
+					debug_assert!(startup_complete);
 					self.handle_monitor_update_release(counterparty_node_id, channel_id, None);
 				},
+				EventCompletionAction::ReleasePaymentCompleteChannelMonitorUpdate(
+					PaymentCompleteUpdate {
+						counterparty_node_id,
+						channel_funding_outpoint,
+						channel_id,
+						htlc_id,
+					},
+				) => {
+					let per_peer_state = self.per_peer_state.read().unwrap();
+					let mut peer_state = per_peer_state
+						.get(&counterparty_node_id)
+						.map(|state| state.lock().unwrap())
+						.expect("Channels originating a payment resolution must have peer state");
+					let update_id = peer_state
+						.closed_channel_monitor_update_ids
+						.get_mut(&channel_id)
+						.expect("Channels originating a payment resolution must have a monitor");
+					*update_id += 1;
+
+					let update = ChannelMonitorUpdate {
+						update_id: *update_id,
+						channel_id: Some(channel_id),
+						updates: vec![ChannelMonitorUpdateStep::ReleasePaymentComplete {
+							htlc: htlc_id,
+						}],
+					};
+
+					let during_startup =
+						!self.background_events_processed_since_startup.load(Ordering::Acquire);
+					if during_startup {
+						let event = BackgroundEvent::MonitorUpdateRegeneratedOnStartup {
+							counterparty_node_id,
+							funding_txo: channel_funding_outpoint,
+							channel_id,
+							update,
+						};
+						self.pending_background_events.lock().unwrap().push(event);
+					} else {
+						handle_new_monitor_update!(
+							self,
+							channel_funding_outpoint,
+							update,
+							peer_state,
+							peer_state,
+							per_peer_state,
+							counterparty_node_id,
+							channel_id,
+							POST_CHANNEL_CLOSE
+						);
+					}
+				},
 			}
 		}
 	}
@@ -16557,6 +16648,7 @@ where
 							monitor,
 							Some(htlc.payment_hash),
 						);
+						let htlc_id = SentHTLCId::from_source(&htlc_source);
 						match htlc_source {
 							HTLCSource::PreviousHopData(prev_hop_data) => {
 								let pending_forward_matches_htlc = |info: &PendingAddHTLCInfo| {
@@ -16614,6 +16706,15 @@ where
 							} => {
 								if let Some(preimage) = preimage_opt {
 									let pending_events = Mutex::new(pending_events_read);
+									let update = PaymentCompleteUpdate {
+										counterparty_node_id: monitor.get_counterparty_node_id(),
+										channel_funding_outpoint: monitor.get_funding_txo(),
+										channel_id: monitor.channel_id(),
+										htlc_id,
+									};
+									let mut compl_action = Some(
+										EventCompletionAction::ReleasePaymentCompleteChannelMonitorUpdate(update)
+									);
 									// Note that we set `from_onchain` to "false" here,
 									// deliberately keeping the pending payment around forever.
 									// Given it should only occur when we have a channel we're
@@ -16622,15 +16723,6 @@ where
 									// generating a `PaymentPathSuccessful` event but regenerating
 									// it and the `PaymentSent` on every restart until the
 									// `ChannelMonitor` is removed.
-									let mut compl_action = Some(
-										EventCompletionAction::ReleaseRAAChannelMonitorUpdate {
-											channel_funding_outpoint: Some(
-												monitor.get_funding_txo(),
-											),
-											channel_id: monitor.channel_id(),
-											counterparty_node_id: path.hops[0].pubkey,
-										},
-									);
 									pending_outbounds.claim_htlc(
 										payment_id,
 										preimage,
@@ -16642,6 +16734,41 @@ where
 										&pending_events,
 										&&logger,
 									);
+									// If the completion action was not consumed, then there was no
+									// payment to claim, and we need to tell the `ChannelMonitor`
+									// we don't need to hear about the HTLC again, at least as long
+									// as the PaymentSent event isn't still sitting around in our
+									// event queue.
+									let have_action = if compl_action.is_some() {
+										let pending_events = pending_events.lock().unwrap();
+										pending_events.iter().any(|(_, act)| *act == compl_action)
+									} else {
+										false
+									};
+									if !have_action && compl_action.is_some() {
+										let mut peer_state = per_peer_state
+											.get(&counterparty_node_id)
+											.map(|state| state.lock().unwrap())
+											.expect("Channels originating a preimage must have peer state");
+										let update_id = peer_state
+											.closed_channel_monitor_update_ids
+											.get_mut(channel_id)
+											.expect("Channels originating a preimage must have a monitor");
+										*update_id += 1;
+
+										pending_background_events.push(BackgroundEvent::MonitorUpdateRegeneratedOnStartup {
+											counterparty_node_id: monitor.get_counterparty_node_id(),
+											funding_txo: monitor.get_funding_txo(),
+											channel_id: monitor.channel_id(),
+											update: ChannelMonitorUpdate {
+												update_id: *update_id,
+												channel_id: Some(monitor.channel_id()),
+												updates: vec![ChannelMonitorUpdateStep::ReleasePaymentComplete {
+													htlc: htlc_id,
+												}],
+											},
+										});
+									}
 									pending_events_read = pending_events.into_inner().unwrap();
 								}
 							},
diff --git a/lightning/src/ln/functional_test_utils.rs b/lightning/src/ln/functional_test_utils.rs
index 2c839c4..a27d347 100644
--- a/lightning/src/ln/functional_test_utils.rs
+++ b/lightning/src/ln/functional_test_utils.rs
@@ -2803,6 +2803,9 @@ pub fn expect_payment_sent<CM: AChannelManager, H: NodeHolder<CM = CM>>(
 	expected_fee_msat_opt: Option<Option<u64>>, expect_per_path_claims: bool,
 	expect_post_ev_mon_update: bool,
 ) -> (Option<PaidBolt12Invoice>, Vec<Event>) {
+	if expect_post_ev_mon_update {
+		check_added_monitors(node, 0);
+	}
 	let events = node.node().get_and_clear_pending_events();
 	let expected_payment_hash = PaymentHash(
 		bitcoin::hashes::sha256::Hash::hash(&expected_payment_preimage.0).to_byte_array(),
@@ -3052,6 +3055,7 @@ pub struct PaymentFailedConditions<'a> {
 	pub(crate) expected_blamed_chan_closed: Option<bool>,
 	pub(crate) expected_mpp_parts_remain: bool,
 	pub(crate) retry_expected: bool,
+	pub(crate) from_mon_update: bool,
 }
 
 impl<'a> PaymentFailedConditions<'a> {
@@ -3062,6 +3066,7 @@ impl<'a> PaymentFailedConditions<'a> {
 			expected_blamed_chan_closed: None,
 			expected_mpp_parts_remain: false,
 			retry_expected: false,
+			from_mon_update: false,
 		}
 	}
 	pub fn mpp_parts_remain(mut self) -> Self {
@@ -3086,6 +3091,10 @@ impl<'a> PaymentFailedConditions<'a> {
 		self.retry_expected = true;
 		self
 	}
+	pub fn from_mon_update(mut self) -> Self {
+		self.from_mon_update = true;
+		self
+	}
 }
 
 #[cfg(any(test, feature = "_externalize_tests"))]
@@ -3195,7 +3204,13 @@ pub fn expect_payment_failed_conditions<'a, 'b, 'c, 'd, 'e>(
 	node: &'a Node<'b, 'c, 'd>, expected_payment_hash: PaymentHash,
 	expected_payment_failed_permanently: bool, conditions: PaymentFailedConditions<'e>,
 ) {
+	if conditions.from_mon_update {
+		check_added_monitors(node, 0);
+	}
 	let events = node.node.get_and_clear_pending_events();
+	if conditions.from_mon_update {
+		check_added_monitors(node, 1);
+	}
 	expect_payment_failed_conditions_event(
 		events,
 		expected_payment_hash,
diff --git a/lightning/src/ln/functional_tests.rs b/lightning/src/ln/functional_tests.rs
index 5e78049..599cc6d 100644
--- a/lightning/src/ln/functional_tests.rs
+++ b/lightning/src/ln/functional_tests.rs
@@ -1140,7 +1140,8 @@ pub fn claim_htlc_outputs() {
 		// ANTI_REORG_DELAY confirmations.
 		mine_transaction(&nodes[1], accepted_claim);
 		connect_blocks(&nodes[1], ANTI_REORG_DELAY - 1);
-		expect_payment_failed!(nodes[1], payment_hash_2, false);
+		let conditions = PaymentFailedConditions::new().from_mon_update();
+		expect_payment_failed_conditions(&nodes[1], payment_hash_2, false, conditions);
 	}
 	assert_eq!(nodes[0].node.list_channels().len(), 0);
 	assert_eq!(nodes[1].node.list_channels().len(), 0);
@@ -1621,6 +1622,7 @@ pub fn test_htlc_on_chain_success() {
 	check_closed_broadcast!(nodes[0], true);
 	check_added_monitors(&nodes[0], 1);
 	let events = nodes[0].node.get_and_clear_pending_events();
+	check_added_monitors(&nodes[0], 2);
 	assert_eq!(events.len(), 5);
 	let mut first_claimed = false;
 	for event in events {
@@ -2038,7 +2040,11 @@ fn do_test_commitment_revoked_fail_backward_exhaustive(
 	mine_transaction(&nodes[1], &revoked_local_txn[0]);
 	connect_blocks(&nodes[1], ANTI_REORG_DELAY - 1);
 
+	check_added_monitors(&nodes[1], 0);
 	let events = nodes[1].node.get_and_clear_pending_events();
+	if deliver_bs_raa {
+		check_added_monitors(&nodes[1], 1);
+	}
 	assert_eq!(events.len(), if deliver_bs_raa { 3 + nodes.len() - 1 } else { 3 + nodes.len() });
 	assert!(events.iter().any(|ev| matches!(
 		ev,
@@ -3641,7 +3647,8 @@ pub fn test_static_spendable_outputs_timeout_tx() {
 	mine_transaction(&nodes[1], &node_txn[0]);
 	check_closed_event!(nodes[1], 1, ClosureReason::CommitmentTxConfirmed, [node_a_id], 100000);
 	connect_blocks(&nodes[1], ANTI_REORG_DELAY - 1);
-	expect_payment_failed!(nodes[1], our_payment_hash, false);
+	let conditions = PaymentFailedConditions::new().from_mon_update();
+	expect_payment_failed_conditions(&nodes[1], our_payment_hash, false, conditions);
 
 	let spend_txn = check_spendable_outputs!(nodes[1], node_cfgs[1].keys_manager);
 	assert_eq!(spend_txn.len(), 3); // SpendableOutput: remote_commitment_tx.to_remote, timeout_tx.output
@@ -4713,7 +4720,8 @@ pub fn test_dynamic_spendable_outputs_local_htlc_timeout_tx() {
 
 	mine_transaction(&nodes[0], &htlc_timeout);
 	connect_blocks(&nodes[0], BREAKDOWN_TIMEOUT as u32 - 1);
-	expect_payment_failed!(nodes[0], our_payment_hash, false);
+	let conditions = PaymentFailedConditions::new().from_mon_update();
+	expect_payment_failed_conditions(&nodes[0], our_payment_hash, false, conditions);
 
 	// Verify that A is able to spend its own HTLC-Timeout tx thanks to spendable output event given back by its ChannelMonitor
 	let spend_txn = check_spendable_outputs!(nodes[0], node_cfgs[0].keys_manager);
@@ -4833,7 +4841,8 @@ pub fn test_key_derivation_params() {
 
 	mine_transaction(&nodes[0], &htlc_timeout);
 	connect_blocks(&nodes[0], BREAKDOWN_TIMEOUT as u32 - 1);
-	expect_payment_failed!(nodes[0], our_payment_hash, false);
+	let conditions = PaymentFailedConditions::new().from_mon_update();
+	expect_payment_failed_conditions(&nodes[0], our_payment_hash, false, conditions);
 
 	// Verify that A is able to spend its own HTLC-Timeout tx thanks to spendable output event given back by its ChannelMonitor
 	let new_keys_manager = test_utils::TestKeysInterface::new(&seed, Network::Testnet);
@@ -5739,7 +5748,9 @@ fn do_test_failure_delay_dust_htlc_local_commitment(announce_latest: bool) {
 
 	assert_eq!(nodes[0].node.get_and_clear_pending_events().len(), 0);
 	connect_blocks(&nodes[0], ANTI_REORG_DELAY - 1);
+	check_added_monitors(&nodes[0], 0);
 	let events = nodes[0].node.get_and_clear_pending_events();
+	check_added_monitors(&nodes[0], 2);
 	// Only 2 PaymentPathFailed events should show up, over-dust HTLC has to be failed by timeout tx
 	assert_eq!(events.len(), 4);
 	let mut first_failed = false;
@@ -5808,12 +5819,14 @@ fn do_test_sweep_outbound_htlc_failure_update(revoked: bool, local: bool) {
 		// We fail dust-HTLC 1 by broadcast of local commitment tx
 		mine_transaction(&nodes[0], &as_commitment_tx[0]);
 		check_closed_event!(nodes[0], 1, ClosureReason::CommitmentTxConfirmed, [node_b_id], 100000);
+		check_closed_broadcast!(nodes[0], true);
+		check_added_monitors(&nodes[0], 1);
 		connect_blocks(&nodes[0], ANTI_REORG_DELAY - 1);
-		expect_payment_failed!(nodes[0], dust_hash, false);
+		let conditions = PaymentFailedConditions::new().from_mon_update();
+		expect_payment_failed_conditions(&nodes[0], dust_hash, false, conditions);
 
 		connect_blocks(&nodes[0], TEST_FINAL_CLTV + LATENCY_GRACE_PERIOD_BLOCKS - ANTI_REORG_DELAY);
-		check_closed_broadcast!(nodes[0], true);
-		check_added_monitors(&nodes[0], 1);
+		check_added_monitors(&nodes[0], 0);
 		assert_eq!(nodes[0].node.get_and_clear_pending_events().len(), 0);
 		timeout_tx.push(nodes[0].tx_broadcaster.txn_broadcasted.lock().unwrap()[0].clone());
 		assert_eq!(
@@ -5824,7 +5837,8 @@ fn do_test_sweep_outbound_htlc_failure_update(revoked: bool, local: bool) {
 		assert_eq!(nodes[0].node.get_and_clear_pending_events().len(), 0);
 		mine_transaction(&nodes[0], &timeout_tx[0]);
 		connect_blocks(&nodes[0], ANTI_REORG_DELAY - 1);
-		expect_payment_failed!(nodes[0], non_dust_hash, false);
+		let conditions = PaymentFailedConditions::new().from_mon_update();
+		expect_payment_failed_conditions(&nodes[0], non_dust_hash, false, conditions);
 	} else {
 		// We fail dust-HTLC 1 by broadcast of remote commitment tx. If revoked, fail also non-dust HTLC
 		mine_transaction(&nodes[0], &bs_commitment_tx[0]);
@@ -5845,7 +5859,8 @@ fn do_test_sweep_outbound_htlc_failure_update(revoked: bool, local: bool) {
 		check_spends!(timeout_tx[0], bs_commitment_tx[0]);
 		// For both a revoked or non-revoked commitment transaction, after ANTI_REORG_DELAY the
 		// dust HTLC should have been failed.
-		expect_payment_failed!(nodes[0], dust_hash, false);
+		let conditions = PaymentFailedConditions::new().from_mon_update();
+		expect_payment_failed_conditions(&nodes[0], dust_hash, false, conditions);
 
 		if !revoked {
 			assert_eq!(
@@ -5859,7 +5874,8 @@ fn do_test_sweep_outbound_htlc_failure_update(revoked: bool, local: bool) {
 		mine_transaction(&nodes[0], &timeout_tx[0]);
 		assert_eq!(nodes[0].node.get_and_clear_pending_events().len(), 0);
 		connect_blocks(&nodes[0], ANTI_REORG_DELAY - 1);
-		expect_payment_failed!(nodes[0], non_dust_hash, false);
+		let conditions = PaymentFailedConditions::new().from_mon_update();
+		expect_payment_failed_conditions(&nodes[0], non_dust_hash, false, conditions);
 	}
 }
 
@@ -7548,7 +7564,8 @@ pub fn test_htlc_no_detection() {
 		&create_dummy_block(nodes[0].best_block_hash(), 42, vec![htlc_timeout.clone()]),
 	);
 	connect_blocks(&nodes[0], ANTI_REORG_DELAY - 1);
-	expect_payment_failed!(nodes[0], our_payment_hash, false);
+	let conditions = PaymentFailedConditions::new().from_mon_update();
+	expect_payment_failed_conditions(&nodes[0], our_payment_hash, false, conditions);
 }
 
 fn do_test_onchain_htlc_settlement_after_close(
diff --git a/lightning/src/ln/monitor_tests.rs b/lightning/src/ln/monitor_tests.rs
index 8e08e5c..03c9e2b 100644
--- a/lightning/src/ln/monitor_tests.rs
+++ b/lightning/src/ln/monitor_tests.rs
@@ -165,7 +165,8 @@ fn revoked_output_htlc_resolution_timing() {
 	assert!(nodes[1].node.get_and_clear_pending_events().is_empty());
 
 	connect_blocks(&nodes[1], ANTI_REORG_DELAY - 1);
-	expect_payment_failed!(nodes[1], payment_hash_1, false);
+	let conditions = PaymentFailedConditions::new().from_mon_update();
+	expect_payment_failed_conditions(&nodes[1], payment_hash_1, false, conditions);
 }
 
 #[test]
@@ -272,7 +273,7 @@ fn archive_fully_resolved_monitors() {
 
 	// Finally, we process the pending `MonitorEvent` from nodes[0], allowing the `ChannelMonitor`
 	// to be archived `ARCHIVAL_DELAY_BLOCKS` blocks later.
-	expect_payment_sent(&nodes[0], payment_preimage, None, true, false);
+	expect_payment_sent(&nodes[0], payment_preimage, None, true, true);
 	nodes[0].chain_monitor.chain_monitor.archive_fully_resolved_channel_monitors();
 	assert_eq!(nodes[0].chain_monitor.chain_monitor.list_monitors().len(), 1);
 	connect_blocks(&nodes[0], ARCHIVAL_DELAY_BLOCKS - 1);
@@ -704,7 +705,8 @@ fn do_test_claim_value_force_close(anchors: bool, prev_commitment_tx: bool) {
 		sorted_vec(nodes[1].chain_monitor.chain_monitor.get_monitor(chan_id).unwrap().get_claimable_balances()));
 
 	connect_blocks(&nodes[0], ANTI_REORG_DELAY - 1);
-	expect_payment_failed!(nodes[0], dust_payment_hash, false);
+	let conditions = PaymentFailedConditions::new().from_mon_update();
+	expect_payment_failed_conditions(&nodes[0], dust_payment_hash, false, conditions);
 	connect_blocks(&nodes[1], ANTI_REORG_DELAY - 1);
 
 	// After ANTI_REORG_DELAY, A will consider its balance fully spendable and generate a
@@ -727,8 +729,9 @@ fn do_test_claim_value_force_close(anchors: bool, prev_commitment_tx: bool) {
 	mine_transaction(&nodes[0], &b_broadcast_txn[0]);
 	if prev_commitment_tx {
 		expect_payment_path_successful!(nodes[0]);
+		check_added_monitors(&nodes[0], 1);
 	} else {
-		expect_payment_sent(&nodes[0], payment_preimage, None, true, false);
+		expect_payment_sent(&nodes[0], payment_preimage, None, true, true);
 	}
 	assert_eq!(sorted_vec(vec![sent_htlc_balance.clone(), sent_htlc_timeout_balance.clone()]),
 		sorted_vec(nodes[0].chain_monitor.chain_monitor.get_monitor(chan_id).unwrap().get_claimable_balances()));
@@ -760,7 +763,8 @@ fn do_test_claim_value_force_close(anchors: bool, prev_commitment_tx: bool) {
 	connect_blocks(&nodes[0], ANTI_REORG_DELAY - 1);
 	assert_eq!(Vec::<Balance>::new(),
 		nodes[0].chain_monitor.chain_monitor.get_monitor(chan_id).unwrap().get_claimable_balances());
-	expect_payment_failed!(nodes[0], timeout_payment_hash, false);
+	let conditions = PaymentFailedConditions::new().from_mon_update();
+	expect_payment_failed_conditions(&nodes[0], timeout_payment_hash, false, conditions);
 
 	test_spendable_output(&nodes[0], &a_htlc_timeout_tx, false);
 
@@ -984,7 +988,7 @@ fn do_test_balances_on_local_commitment_htlcs(anchors: bool) {
 	// Now confirm nodes[1]'s HTLC claim, giving nodes[0] the preimage. Note that the "maybe
 	// claimable" balance remains until we see ANTI_REORG_DELAY blocks.
 	mine_transaction(&nodes[0], &bs_htlc_claim_txn[0]);
-	expect_payment_sent(&nodes[0], payment_preimage_2, None, true, false);
+	expect_payment_sent(&nodes[0], payment_preimage_2, None, true, true);
 	assert_eq!(sorted_vec(vec![Balance::ClaimableAwaitingConfirmations {
 			amount_satoshis: 1_000_000 - 10_000 - 20_000 - commitment_tx_fee - anchor_outputs_value,
 			confirmation_height: node_a_commitment_claimable,
@@ -1026,7 +1030,8 @@ fn do_test_balances_on_local_commitment_htlcs(anchors: bool) {
 	// panicked as described in the test introduction. This will remove the "maybe claimable"
 	// spendable output as nodes[1] has fully claimed the second HTLC.
 	connect_blocks(&nodes[0], ANTI_REORG_DELAY - 1);
-	expect_payment_failed!(nodes[0], payment_hash, false);
+	let conditions = PaymentFailedConditions::new().from_mon_update();
+	expect_payment_failed_conditions(&nodes[0], payment_hash, false, conditions);
 
 	assert_eq!(sorted_vec(vec![Balance::ClaimableAwaitingConfirmations {
 			amount_satoshis: 1_000_000 - 10_000 - 20_000 - commitment_tx_fee - anchor_outputs_value,
@@ -1256,7 +1261,8 @@ fn test_no_preimage_inbound_htlc_balances() {
 	// Once as_htlc_timeout_claim[0] reaches ANTI_REORG_DELAY confirmations, we should get a
 	// payment failure event.
 	connect_blocks(&nodes[0], ANTI_REORG_DELAY - 2);
-	expect_payment_failed!(nodes[0], to_b_failed_payment_hash, false);
+	let conditions = PaymentFailedConditions::new().from_mon_update();
+	expect_payment_failed_conditions(&nodes[0], to_b_failed_payment_hash, false, conditions);
 
 	connect_blocks(&nodes[0], 1);
 	assert_eq!(sorted_vec(vec![Balance::ClaimableAwaitingConfirmations {
@@ -1304,7 +1310,8 @@ fn test_no_preimage_inbound_htlc_balances() {
 		sorted_vec(nodes[1].chain_monitor.chain_monitor.get_monitor(chan_id).unwrap().get_claimable_balances()));
 
 	connect_blocks(&nodes[1], ANTI_REORG_DELAY - 2);
-	expect_payment_failed!(nodes[1], to_a_failed_payment_hash, false);
+	let conditions = PaymentFailedConditions::new().from_mon_update();
+	expect_payment_failed_conditions(&nodes[1], to_a_failed_payment_hash, false, conditions);
 
 	assert_eq!(vec![b_received_htlc_balance.clone()],
 		nodes[1].chain_monitor.chain_monitor.get_monitor(chan_id).unwrap().get_claimable_balances());
@@ -1572,7 +1579,9 @@ fn do_test_revoked_counterparty_commitment_balances(anchors: bool, confirm_htlc_
 	connect_blocks(&nodes[1], 3);
 	test_spendable_output(&nodes[1], &as_revoked_txn[0], false);
 
+	check_added_monitors(&nodes[1], 0);
 	let mut payment_failed_events = nodes[1].node.get_and_clear_pending_events();
+	check_added_monitors(&nodes[1], 2);
 	expect_payment_failed_conditions_event(payment_failed_events[..2].to_vec(),
 		missing_htlc_payment_hash, false, PaymentFailedConditions::new());
 	expect_payment_failed_conditions_event(payment_failed_events[2..].to_vec(),
@@ -1581,7 +1590,9 @@ fn do_test_revoked_counterparty_commitment_balances(anchors: bool, confirm_htlc_
 	connect_blocks(&nodes[1], 1);
 	if confirm_htlc_spend_first {
 		test_spendable_output(&nodes[1], &claim_txn[0], false);
+		check_added_monitors(&nodes[1], 0);
 		let mut payment_failed_events = nodes[1].node.get_and_clear_pending_events();
+		check_added_monitors(&nodes[1], 2);
 		expect_payment_failed_conditions_event(payment_failed_events[..2].to_vec(),
 			live_payment_hash, false, PaymentFailedConditions::new());
 		expect_payment_failed_conditions_event(payment_failed_events[2..].to_vec(),
@@ -1594,7 +1605,9 @@ fn do_test_revoked_counterparty_commitment_balances(anchors: bool, confirm_htlc_
 		test_spendable_output(&nodes[1], &claim_txn[1], false);
 	} else {
 		test_spendable_output(&nodes[1], &claim_txn[0], false);
+		check_added_monitors(&nodes[1], 0);
 		let mut payment_failed_events = nodes[1].node.get_and_clear_pending_events();
+		check_added_monitors(&nodes[1], 2);
 		expect_payment_failed_conditions_event(payment_failed_events[..2].to_vec(),
 			live_payment_hash, false, PaymentFailedConditions::new());
 		expect_payment_failed_conditions_event(payment_failed_events[2..].to_vec(),
@@ -2042,7 +2055,7 @@ fn do_test_revoked_counterparty_aggregated_claims(anchors: bool) {
 		as_revoked_txn[1].clone()
 	};
 	mine_transaction(&nodes[1], &htlc_success_claim);
-	expect_payment_sent(&nodes[1], claimed_payment_preimage, None, true, false);
+	expect_payment_sent(&nodes[1], claimed_payment_preimage, None, true, true);
 
 	let mut claim_txn_2 = nodes[1].tx_broadcaster.txn_broadcast();
 	// Once B sees the HTLC-Success transaction it splits its claim transaction into two, though in
@@ -2143,7 +2156,8 @@ fn do_test_revoked_counterparty_aggregated_claims(anchors: bool) {
 	assert!(nodes[1].node.get_and_clear_pending_events().is_empty()); // We shouldn't fail the payment until we spend the output
 
 	connect_blocks(&nodes[1], 5);
-	expect_payment_failed!(nodes[1], revoked_payment_hash, false);
+	let conditions = PaymentFailedConditions::new().from_mon_update();
+	expect_payment_failed_conditions(&nodes[1], revoked_payment_hash, false, conditions);
 	let spendable_output_events = nodes[1].chain_monitor.chain_monitor.get_and_clear_pending_events();
 	assert_eq!(spendable_output_events.len(), 2);
 	for event in spendable_output_events {
@@ -2619,7 +2633,8 @@ fn do_test_yield_anchors_events(have_htlcs: bool) {
 	connect_blocks(&nodes[0], ANTI_REORG_DELAY - 1);
 
 	assert!(nodes[0].chain_monitor.chain_monitor.get_and_clear_pending_events().is_empty());
-	expect_payment_failed!(nodes[0], payment_hash_1.unwrap(), false);
+	let conditions = PaymentFailedConditions::new().from_mon_update();
+	expect_payment_failed_conditions(&nodes[0], payment_hash_1.unwrap(), false, conditions);
 
 	connect_blocks(&nodes[0], BREAKDOWN_TIMEOUT as u32);
 
@@ -3433,6 +3448,7 @@ fn do_test_lost_preimage_monitor_events(on_counterparty_tx: bool) {
 	let mons = &[&mon_a_ser[..], &mon_b_ser[..]];
 	reload_node!(nodes[1], cfg, &node_ser, mons, persister, new_chain_mon, node_b_reload);
 
+	check_added_monitors(&nodes[1], 0);
 	let preimage_events = nodes[1].node.get_and_clear_pending_events();
 	assert_eq!(preimage_events.len(), 2, "{preimage_events:?}");
 	for ev in preimage_events {
@@ -3450,7 +3466,9 @@ fn do_test_lost_preimage_monitor_events(on_counterparty_tx: bool) {
 	// After the background events are processed in `get_and_clear_pending_events`, above, node B
 	// will create the requisite `ChannelMontiorUpdate` for claiming the forwarded payment back.
 	// The HTLC, however, is added to the holding cell for replay after the peer connects, below.
-	check_added_monitors(&nodes[1], 1);
+	// It will also apply a `ChannelMonitorUpdate` to let the `ChannelMonitor` know that the
+	// payment can now be forgotten as the `PaymentSent` event was handled.
+	check_added_monitors(&nodes[1], 2);
 
 	nodes[0].node.peer_disconnected(nodes[1].node.get_our_node_id());
 
@@ -3669,7 +3687,11 @@ fn do_test_lost_timeout_monitor_events(confirm_tx: CommitmentType, dust_htlcs: b
 	let mons = &[&mon_a_ser[..], &mon_b_ser[..]];
 	reload_node!(nodes[1], cfg, &node_ser, mons, persister, new_chain_mon, node_b_reload);
 
+	// After reload, once we process the `PaymentFailed` event, the sent HTLC will be marked
+	// handled so that we won't ever see the event again.
+	check_added_monitors(&nodes[1], 0);
 	let timeout_events = nodes[1].node.get_and_clear_pending_events();
+	check_added_monitors(&nodes[1], 1);
 	assert_eq!(timeout_events.len(), 3, "{timeout_events:?}");
 	for ev in timeout_events {
 		match ev {
diff --git a/lightning/src/ln/outbound_payment.rs b/lightning/src/ln/outbound_payment.rs
index 373536d..caf7ce7 100644
--- a/lightning/src/ln/outbound_payment.rs
+++ b/lightning/src/ln/outbound_payment.rs
@@ -2185,8 +2185,6 @@ impl OutboundPayments {
 				// This could potentially lead to removing a pending payment too early,
 				// with a reorg of one block causing us to re-add the fulfilled payment on
 				// restart.
-				// TODO: We should have a second monitor event that informs us of payments
-				// irrevocably fulfilled.
 				if payment.get_mut().remove(&session_priv_bytes, Some(&path)) {
 					let payment_hash = Some(PaymentHash(Sha256::hash(&payment_preimage.0).to_byte_array()));
 					pending_events.push_back((events::Event::PaymentPathSuccessful {
@@ -2474,10 +2472,14 @@ impl OutboundPayments {
 			}
 		};
 		let mut pending_events = pending_events.lock().unwrap();
-		// TODO: Handle completion_action
-		pending_events.push_back((path_failure, None));
+		let completion_action = completion_action
+			.take()
+			.map(|act| EventCompletionAction::ReleasePaymentCompleteChannelMonitorUpdate(act));
 		if let Some(ev) = full_failure_ev {
-			pending_events.push_back((ev, None));
+			pending_events.push_back((path_failure, None));
+			pending_events.push_back((ev, completion_action));
+		} else {
+			pending_events.push_back((path_failure, completion_action));
 		}
 	}
 
diff --git a/lightning/src/ln/payment_tests.rs b/lightning/src/ln/payment_tests.rs
index 0af0463..1b3749d 100644
--- a/lightning/src/ln/payment_tests.rs
+++ b/lightning/src/ln/payment_tests.rs
@@ -933,7 +933,7 @@ fn do_retry_with_no_persist(confirm_before_reload: bool) {
 		assert_eq!(txn[0].compute_txid(), as_commitment_tx.compute_txid());
 	}
 	mine_transaction(&nodes[0], &bs_htlc_claim_txn);
-	expect_payment_sent(&nodes[0], payment_preimage_1, None, true, false);
+	expect_payment_sent(&nodes[0], payment_preimage_1, None, true, true);
 	connect_blocks(&nodes[0], TEST_FINAL_CLTV * 4 + 20);
 	let (first_htlc_timeout_tx, second_htlc_timeout_tx) = {
 		let mut txn = nodes[0].tx_broadcaster.unique_txn_broadcast();
@@ -949,7 +949,7 @@ fn do_retry_with_no_persist(confirm_before_reload: bool) {
 		confirm_transaction(&nodes[0], &first_htlc_timeout_tx);
 	}
 	nodes[0].tx_broadcaster.txn_broadcasted.lock().unwrap().clear();
-	let conditions = PaymentFailedConditions::new();
+	let conditions = PaymentFailedConditions::new().from_mon_update();
 	expect_payment_failed_conditions(&nodes[0], payment_hash, false, conditions);
 
 	// Finally, retry the payment (which was reloaded from the ChannelMonitor when nodes[0] was
@@ -1164,7 +1164,8 @@ fn do_test_completed_payment_not_retryable_on_reload(use_dust: bool) {
 	// (which should also still work).
 	connect_blocks(&nodes[0], ANTI_REORG_DELAY - 1);
 	connect_blocks(&nodes[1], ANTI_REORG_DELAY - 1);
-	expect_payment_failed_conditions(&nodes[0], hash, false, PaymentFailedConditions::new());
+	let conditions = PaymentFailedConditions::new().from_mon_update();
+	expect_payment_failed_conditions(&nodes[0], hash, false, conditions);
 
 	let chan_0_monitor_serialized = get_monitor!(nodes[0], chan_id).encode();
 	let chan_1_monitor_serialized = get_monitor!(nodes[0], chan_id_3).encode();
@@ -1181,6 +1182,9 @@ fn do_test_completed_payment_not_retryable_on_reload(use_dust: bool) {
 	nodes[1].node.peer_disconnected(node_a_id);
 
 	nodes[0].node.test_process_background_events();
+	check_added_monitors(&nodes[0], 1); // TODO: Removed in the next commit as this only required
+									// when we are still seeing all payments, even resolved
+									// ones.
 
 	let mut reconnect_args = ReconnectArgs::new(&nodes[0], &nodes[1]);
 	reconnect_args.send_channel_ready = (true, true);
@@ -1213,6 +1217,9 @@ fn do_test_completed_payment_not_retryable_on_reload(use_dust: bool) {
 	nodes[1].node.peer_disconnected(node_a_id);
 
 	nodes[0].node.test_process_background_events();
+	check_added_monitors(&nodes[0], 1); // TODO: Removed in the next commit as this only required
+									// when we are still seeing all payments, even resolved
+									// ones.
 
 	reconnect_nodes(ReconnectArgs::new(&nodes[0], &nodes[1]));
 
@@ -1333,9 +1340,10 @@ fn do_test_dup_htlc_onchain_doesnt_fail_on_reload(
 
 	let mon_ser = get_monitor!(nodes[0], chan_id).encode();
 	if payment_timeout {
-		expect_payment_failed!(nodes[0], payment_hash, false);
+		let conditions = PaymentFailedConditions::new().from_mon_update();
+		expect_payment_failed_conditions(&nodes[0], payment_hash, false, conditions);
 	} else {
-		expect_payment_sent(&nodes[0], payment_preimage, None, true, false);
+		expect_payment_sent(&nodes[0], payment_preimage, None, true, true);
 	}
 
 	// If we persist the ChannelManager after we get the PaymentSent event, we shouldn't get it
@@ -1347,12 +1355,20 @@ fn do_test_dup_htlc_onchain_doesnt_fail_on_reload(
 	// Now reload nodes[0]...
 	reload_node!(nodes[0], &node_a_ser, &[&mon_ser], persister, chain_monitor, node_a_reload);
 
+	check_added_monitors(&nodes[0], 0);
 	if persist_manager_post_event {
 		assert!(nodes[0].node.get_and_clear_pending_events().is_empty());
+		check_added_monitors(&nodes[0], 2);
 	} else if payment_timeout {
-		expect_payment_failed!(nodes[0], payment_hash, false);
+		let conditions = PaymentFailedConditions::new().from_mon_update();
+		expect_payment_failed_conditions(&nodes[0], payment_hash, false, conditions);
 	} else {
+		// After reload, the ChannelManager identified the failed payment and queued up the
+		// PaymentSent and corresponding ChannelMonitorUpdate to mark the payment handled, but
+		// while processing the pending `MonitorEvent`s (which were not processed before the
+		// monitor was persisted) we will end up with a duplicate ChannelMonitorUpdate.
 		expect_payment_sent(&nodes[0], payment_preimage, None, true, false);
+		check_added_monitors(&nodes[0], 2);
 	}
 
 	// Note that if we re-connect the block which exposed nodes[0] to the payment preimage (but
@@ -1625,7 +1641,9 @@ fn onchain_failed_probe_yields_event() {
 	check_closed_broadcast!(&nodes[0], true);
 	check_added_monitors!(nodes[0], 1);
 
+	check_added_monitors(&nodes[0], 0);
 	let mut events = nodes[0].node.get_and_clear_pending_events();
+	check_added_monitors(&nodes[0], 1);
 	assert_eq!(events.len(), 2);
 	let mut found_probe_failed = false;
 	for event in events.drain(..) {
@@ -4084,7 +4102,14 @@ fn do_no_missing_sent_on_reload(persist_manager_with_payment: bool, at_midpoint:
 	let config = test_default_channel_config();
 	reload_node!(nodes[0], config, &node_a_ser, &[&mon_ser], persist_a, chain_monitor_a, node_a_1);
 
+	// When we first process background events, we'll apply a channel-closed monitor update...
+	check_added_monitors(&nodes[0], 0);
+	nodes[0].node.test_process_background_events();
+	check_added_monitors(&nodes[0], 1);
+	// Then once we process the PaymentSent event we'll apply a monitor update to remove the
+	// pending payment from being re-hydrated on the next startup.
 	let events = nodes[0].node.get_and_clear_pending_events();
+	check_added_monitors(&nodes[0], 1);
 	assert_eq!(events.len(), 2);
 	if let Event::ChannelClosed { reason: ClosureReason::OutdatedChannelManager, .. } = events[0] {
 	} else {
@@ -4114,8 +4139,16 @@ fn do_no_missing_sent_on_reload(persist_manager_with_payment: bool, at_midpoint:
 	let node_ser = nodes[0].node.encode();
 	let config = test_default_channel_config();
 	reload_node!(nodes[0], config, &node_ser, &[&mon_ser], persist_b, chain_monitor_b, node_a_2);
+
+	// Because the pending payment will currently stick around forever, we'll apply a
+	// ChannelMonitorUpdate on each startup to attempt to remove it.
+	// TODO: This will be dropped in the next commit after we actually remove the payment!
+	check_added_monitors(&nodes[0], 0);
+	nodes[0].node.test_process_background_events();
+	check_added_monitors(&nodes[0], 1);
 	let events = nodes[0].node.get_and_clear_pending_events();
 	assert!(events.is_empty());
+	check_added_monitors(&nodes[0], 0);
 
 	// Ensure that we don't generate any further events even after the channel-closing commitment
 	// transaction is confirmed on-chain.
@@ -4126,6 +4159,7 @@ fn do_no_missing_sent_on_reload(persist_manager_with_payment: bool, at_midpoint:
 
 	let events = nodes[0].node.get_and_clear_pending_events();
 	assert!(events.is_empty());
+	check_added_monitors(&nodes[0], 0);
 
 	let mon_ser = get_monitor!(nodes[0], chan_id).encode();
 	let config = test_default_channel_config();
@@ -4133,6 +4167,9 @@ fn do_no_missing_sent_on_reload(persist_manager_with_payment: bool, at_midpoint:
 	reload_node!(nodes[0], config, &node_ser, &[&mon_ser], persist_c, chain_monitor_c, node_a_3);
 	let events = nodes[0].node.get_and_clear_pending_events();
 	assert!(events.is_empty());
+
+	// TODO: This will be dropped in the next commit after we actually remove the payment!
+	check_added_monitors(&nodes[0], 1);
 }
 
 #[test]
diff --git a/lightning/src/ln/reorg_tests.rs b/lightning/src/ln/reorg_tests.rs
index d85c95c..ddf71a8 100644
--- a/lightning/src/ln/reorg_tests.rs
+++ b/lightning/src/ln/reorg_tests.rs
@@ -237,12 +237,13 @@ fn test_counterparty_revoked_reorg() {
 
 	// Connect the HTLC claim transaction for HTLC 3
 	mine_transaction(&nodes[1], &unrevoked_local_txn[2]);
-	expect_payment_sent(&nodes[1], payment_preimage_3, None, true, false);
+	expect_payment_sent(&nodes[1], payment_preimage_3, None, true, true);
 	assert!(nodes[1].node.get_and_clear_pending_msg_events().is_empty());
 
 	// Connect blocks to confirm the unrevoked commitment transaction
 	connect_blocks(&nodes[1], ANTI_REORG_DELAY - 2);
-	expect_payment_failed!(nodes[1], payment_hash_4, false);
+	let conditions = PaymentFailedConditions::new().from_mon_update();
+	expect_payment_failed_conditions(&nodes[1], payment_hash_4, false, conditions)
 }
 
 fn do_test_unconf_chan(reload_node: bool, reorg_after_reload: bool, use_funding_unconfirmed: bool, connect_style: ConnectStyle) {
@@ -1047,7 +1048,9 @@ fn do_test_split_htlc_expiry_tracking(use_third_htlc: bool, reorg_out: bool) {
 	let mut txn = nodes[0].tx_broadcaster.txn_broadcast();
 	assert_eq!(txn.len(), 0);
 
+	check_added_monitors(&nodes[0], 0);
 	let sent_events = nodes[0].node.get_and_clear_pending_events();
+	check_added_monitors(&nodes[0], 2);
 	assert_eq!(sent_events.len(), 4, "{sent_events:?}");
 	let mut found_expected_events = [false, false, false, false];
 	for event in sent_events {
@@ -1136,7 +1139,9 @@ fn do_test_split_htlc_expiry_tracking(use_third_htlc: bool, reorg_out: bool) {
 		// Connect two more blocks to get `as_third_htlc_spend_tx` to `ANTI_REORG_DELAY` confs.
 		connect_blocks(&nodes[0], 2);
 		if use_third_htlc {
+			check_added_monitors(&nodes[0], 0);
 			let failed_events = nodes[0].node.get_and_clear_pending_events();
+			check_added_monitors(&nodes[0], 1);
 			assert_eq!(failed_events.len(), 2);
 			let mut found_expected_events = [false, false];
 			for event in failed_events {