Stop re-hydrating pending payments once they are fully resolved

diff --git a/lightning/src/chain/channelmonitor.rs b/lightning/src/chain/channelmonitor.rs
index b16bf0c..4f0c4a5 100644
--- a/lightning/src/chain/channelmonitor.rs
+++ b/lightning/src/chain/channelmonitor.rs
@@ -3031,10 +3031,6 @@ impl<Signer: EcdsaChannelSigner> ChannelMonitor<Signer> {
 	/// Gets the set of outbound HTLCs which can be (or have been) resolved by this
 	/// `ChannelMonitor`. This is used to determine if an HTLC was removed from the channel prior
 	/// to the `ChannelManager` having been persisted.
-	///
-	/// This is similar to [`Self::get_pending_or_resolved_outbound_htlcs`] except it includes
-	/// HTLCs which were resolved on-chain (i.e. where the final HTLC resolution was done by an
-	/// event from this `ChannelMonitor`).
 	pub(crate) fn get_all_current_outbound_htlcs(
 		&self,
 	) -> HashMap<HTLCSource, (HTLCOutputInCommitment, Option<PaymentPreimage>)> {
@@ -3047,8 +3043,11 @@ impl<Signer: EcdsaChannelSigner> ChannelMonitor<Signer> {
 				for &(ref htlc, ref source_option) in latest_outpoints.iter() {
 					if let &Some(ref source) = source_option {
 						let htlc_id = SentHTLCId::from_source(source);
-						let preimage_opt = us.counterparty_fulfilled_htlcs.get(&htlc_id).cloned();
-						res.insert((**source).clone(), (htlc.clone(), preimage_opt));
+						if !us.htlcs_resolved_to_user.contains(&htlc_id) {
+							let preimage_opt =
+								us.counterparty_fulfilled_htlcs.get(&htlc_id).cloned();
+							res.insert((**source).clone(), (htlc.clone(), preimage_opt));
+						}
 					}
 				}
 			}
@@ -3104,6 +3103,11 @@ impl<Signer: EcdsaChannelSigner> ChannelMonitor<Signer> {
 						} else {
 							continue;
 						};
+						let htlc_id = SentHTLCId::from_source(source);
+						if us.htlcs_resolved_to_user.contains(&htlc_id) {
+							continue;
+						}
+
 						let confirmed = $htlc_iter.find(|(_, conf_src)| Some(source) == *conf_src);
 						if let Some((confirmed_htlc, _)) = confirmed {
 							let filter = |v: &&IrrevocablyResolvedHTLC| {
@@ -3176,96 +3180,6 @@ impl<Signer: EcdsaChannelSigner> ChannelMonitor<Signer> {
 		res
 	}
 
-	/// Gets the set of outbound HTLCs which are pending resolution in this channel or which were
-	/// resolved with a preimage from our counterparty.
-	///
-	/// This is used to reconstruct pending outbound payments on restart in the ChannelManager.
-	///
-	/// Currently, the preimage is unused, however if it is present in the relevant internal state
-	/// an HTLC is always included even if it has been resolved.
-	#[rustfmt::skip]
-	pub(crate) fn get_pending_or_resolved_outbound_htlcs(&self) -> HashMap<HTLCSource, (HTLCOutputInCommitment, Option<PaymentPreimage>)> {
-		let us = self.inner.lock().unwrap();
-		// We're only concerned with the confirmation count of HTLC transactions, and don't
-		// actually care how many confirmations a commitment transaction may or may not have. Thus,
-		// we look for either a FundingSpendConfirmation event or a funding_spend_confirmed.
-		let confirmed_txid = us.funding_spend_confirmed.or_else(|| {
-			us.onchain_events_awaiting_threshold_conf.iter().find_map(|event| {
-				if let OnchainEvent::FundingSpendConfirmation { .. } = event.event {
-					Some(event.txid)
-				} else { None }
-			})
-		});
-
-		if confirmed_txid.is_none() {
-			// If we have not seen a commitment transaction on-chain (ie the channel is not yet
-			// closed), just get the full set.
-			mem::drop(us);
-			return self.get_all_current_outbound_htlcs();
-		}
-
-		let mut res = new_hash_map();
-		macro_rules! walk_htlcs {
-			($holder_commitment: expr, $htlc_iter: expr) => {
-				for (htlc, source) in $htlc_iter {
-					if us.htlcs_resolved_on_chain.iter().any(|v| v.commitment_tx_output_idx == htlc.transaction_output_index) {
-						// We should assert that funding_spend_confirmed is_some() here, but we
-						// have some unit tests which violate HTLC transaction CSVs entirely and
-						// would fail.
-						// TODO: Once tests all connect transactions at consensus-valid times, we
-						// should assert here like we do in `get_claimable_balances`.
-					} else if htlc.offered == $holder_commitment {
-						// If the payment was outbound, check if there's an HTLCUpdate
-						// indicating we have spent this HTLC with a timeout, claiming it back
-						// and awaiting confirmations on it.
-						let htlc_update_confd = us.onchain_events_awaiting_threshold_conf.iter().any(|event| {
-							if let OnchainEvent::HTLCUpdate { commitment_tx_output_idx: Some(commitment_tx_output_idx), .. } = event.event {
-								// If the HTLC was timed out, we wait for ANTI_REORG_DELAY blocks
-								// before considering it "no longer pending" - this matches when we
-								// provide the ChannelManager an HTLC failure event.
-								Some(commitment_tx_output_idx) == htlc.transaction_output_index &&
-									us.best_block.height >= event.height + ANTI_REORG_DELAY - 1
-							} else if let OnchainEvent::HTLCSpendConfirmation { commitment_tx_output_idx, .. } = event.event {
-								// If the HTLC was fulfilled with a preimage, we consider the HTLC
-								// immediately non-pending, matching when we provide ChannelManager
-								// the preimage.
-								Some(commitment_tx_output_idx) == htlc.transaction_output_index
-							} else { false }
-						});
-						if let Some(source) = source {
-							let counterparty_resolved_preimage_opt =
-								us.counterparty_fulfilled_htlcs.get(&SentHTLCId::from_source(source)).cloned();
-							if !htlc_update_confd || counterparty_resolved_preimage_opt.is_some() {
-								res.insert(source.clone(), (htlc.clone(), counterparty_resolved_preimage_opt));
-							}
-						} else {
-							panic!("Outbound HTLCs should have a source");
-						}
-					}
-				}
-			}
-		}
-
-		let commitment_txid = confirmed_txid.unwrap();
-		let funding_spent = get_confirmed_funding_scope!(us);
-
-		if Some(commitment_txid) == funding_spent.current_counterparty_commitment_txid || Some(commitment_txid) == funding_spent.prev_counterparty_commitment_txid {
-			walk_htlcs!(false, funding_spent.counterparty_claimable_outpoints.get(&commitment_txid).unwrap().iter().filter_map(|(a, b)| {
-				if let &Some(ref source) = b {
-					Some((a, Some(&**source)))
-				} else { None }
-			}));
-		} else if commitment_txid == funding_spent.current_holder_commitment_tx.trust().txid() {
-			walk_htlcs!(true, holder_commitment_htlcs!(us, CURRENT_WITH_SOURCES));
-		} else if let Some(prev_commitment_tx) = &funding_spent.prev_holder_commitment_tx {
-			if commitment_txid == prev_commitment_tx.trust().txid() {
-				walk_htlcs!(true, holder_commitment_htlcs!(us, PREV_WITH_SOURCES).unwrap());
-			}
-		}
-
-		res
-	}
-
 	pub(crate) fn get_stored_preimages(
 		&self,
 	) -> HashMap<PaymentHash, (PaymentPreimage, Vec<PaymentClaimDetails>)> {
diff --git a/lightning/src/ln/channelmanager.rs b/lightning/src/ln/channelmanager.rs
index e76d207..2bf2db5 100644
--- a/lightning/src/ln/channelmanager.rs
+++ b/lightning/src/ln/channelmanager.rs
@@ -16600,8 +16600,7 @@ where
 				}
 
 				if is_channel_closed {
-					for (htlc_source, (htlc, _)) in monitor.get_pending_or_resolved_outbound_htlcs()
-					{
+					for (htlc_source, (htlc, _)) in monitor.get_all_current_outbound_htlcs() {
 						let logger = WithChannelMonitor::from(
 							&args.logger,
 							monitor,
diff --git a/lightning/src/ln/functional_tests.rs b/lightning/src/ln/functional_tests.rs
index 599cc6d..4c48cf6 100644
--- a/lightning/src/ln/functional_tests.rs
+++ b/lightning/src/ln/functional_tests.rs
@@ -2043,6 +2043,8 @@ fn do_test_commitment_revoked_fail_backward_exhaustive(
 	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], 2);
+	} else {
 		check_added_monitors(&nodes[1], 1);
 	}
 	assert_eq!(events.len(), if deliver_bs_raa { 3 + nodes.len() - 1 } else { 3 + nodes.len() });
@@ -2060,7 +2062,7 @@ fn do_test_commitment_revoked_fail_backward_exhaustive(
 	)));
 
 	nodes[1].node.process_pending_htlc_forwards();
-	check_added_monitors(&nodes[1], 2);
+	check_added_monitors(&nodes[1], 1);
 
 	let mut events = nodes[1].node.get_and_clear_pending_msg_events();
 	assert_eq!(events.len(), if deliver_bs_raa { 4 } else { 3 });
diff --git a/lightning/src/ln/payment_tests.rs b/lightning/src/ln/payment_tests.rs
index 1b3749d..f93eb3f 100644
--- a/lightning/src/ln/payment_tests.rs
+++ b/lightning/src/ln/payment_tests.rs
@@ -1182,9 +1182,6 @@ 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);
@@ -1217,9 +1214,6 @@ 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]));
 
@@ -1238,7 +1232,8 @@ fn test_completed_payment_not_retryable_on_reload() {
 }
 
 fn do_test_dup_htlc_onchain_doesnt_fail_on_reload(
-	persist_manager_post_event: bool, confirm_commitment_tx: bool, payment_timeout: bool,
+	persist_manager_post_event: bool, persist_monitor_after_events: bool,
+	confirm_commitment_tx: bool, payment_timeout: bool,
 ) {
 	// When a Channel is closed, any outbound HTLCs which were relayed through it are simply
 	// dropped. From there, the ChannelManager relies on the ChannelMonitor having a copy of the
@@ -1338,36 +1333,58 @@ fn do_test_dup_htlc_onchain_doesnt_fail_on_reload(
 		node_a_ser = nodes[0].node.encode();
 	}
 
-	let mon_ser = get_monitor!(nodes[0], chan_id).encode();
+	let mut mon_ser = Vec::new();
+	if !persist_monitor_after_events {
+		mon_ser = get_monitor!(nodes[0], chan_id).encode();
+	}
 	if payment_timeout {
 		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, true);
 	}
+	// Note that if we persist the monitor before processing the events, above, we'll always get
+	// them replayed on restart no matter what
+	if persist_monitor_after_events {
+		mon_ser = get_monitor!(nodes[0], chan_id).encode();
+	}
 
 	// If we persist the ChannelManager after we get the PaymentSent event, we shouldn't get it
 	// twice.
 	if persist_manager_post_event {
 		node_a_ser = nodes[0].node.encode();
+	} else if persist_monitor_after_events {
+		// Persisting the monitor after the events (resulting in a new monitor being persisted) but
+		// didn't persist the manager will result in an FC, which we don't test here.
+		panic!();
 	}
 
 	// 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 {
+	if persist_manager_post_event && persist_monitor_after_events {
 		assert!(nodes[0].node.get_and_clear_pending_events().is_empty());
-		check_added_monitors(&nodes[0], 2);
+		check_added_monitors(&nodes[0], 0);
 	} else if payment_timeout {
-		let conditions = PaymentFailedConditions::new().from_mon_update();
+		let mut conditions = PaymentFailedConditions::new();
+		if !persist_monitor_after_events {
+			conditions = conditions.from_mon_update();
+		}
 		expect_payment_failed_conditions(&nodes[0], payment_hash, false, conditions);
+		check_added_monitors(&nodes[0], 0);
 	} else {
+		if persist_manager_post_event {
+			assert!(nodes[0].node.get_and_clear_pending_events().is_empty());
+		} else {
+			expect_payment_sent(&nodes[0], payment_preimage, None, true, false);
+		}
 		// 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);
+		// PaymentSent (or not, if `persist_manager_post_event` resulted in us detecting we
+		// already did that) 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.
 		check_added_monitors(&nodes[0], 2);
 	}
 
@@ -1381,12 +1398,15 @@ fn do_test_dup_htlc_onchain_doesnt_fail_on_reload(
 
 #[test]
 fn test_dup_htlc_onchain_doesnt_fail_on_reload() {
-	do_test_dup_htlc_onchain_doesnt_fail_on_reload(true, true, true);
-	do_test_dup_htlc_onchain_doesnt_fail_on_reload(true, true, false);
-	do_test_dup_htlc_onchain_doesnt_fail_on_reload(true, false, false);
-	do_test_dup_htlc_onchain_doesnt_fail_on_reload(false, true, true);
-	do_test_dup_htlc_onchain_doesnt_fail_on_reload(false, true, false);
-	do_test_dup_htlc_onchain_doesnt_fail_on_reload(false, false, false);
+	do_test_dup_htlc_onchain_doesnt_fail_on_reload(true, true, true, true);
+	do_test_dup_htlc_onchain_doesnt_fail_on_reload(true, true, true, false);
+	do_test_dup_htlc_onchain_doesnt_fail_on_reload(true, true, false, false);
+	do_test_dup_htlc_onchain_doesnt_fail_on_reload(true, false, true, true);
+	do_test_dup_htlc_onchain_doesnt_fail_on_reload(true, false, true, false);
+	do_test_dup_htlc_onchain_doesnt_fail_on_reload(true, false, false, false);
+	do_test_dup_htlc_onchain_doesnt_fail_on_reload(false, false, true, true);
+	do_test_dup_htlc_onchain_doesnt_fail_on_reload(false, false, true, false);
+	do_test_dup_htlc_onchain_doesnt_fail_on_reload(false, false, false, false);
 }
 
 #[test]
@@ -4140,15 +4160,9 @@ 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_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.
@@ -4167,9 +4181,6 @@ 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]