ln/refactor: extract channelmonitor recovery to external helper

diff --git a/lightning/src/ln/channelmanager.rs b/lightning/src/ln/channelmanager.rs
index 5ac5c0d..f0aaac0 100644
--- a/lightning/src/ln/channelmanager.rs
+++ b/lightning/src/ln/channelmanager.rs
@@ -19244,21 +19244,6 @@ impl<
 			(ChannelId, PaymentHash),
 			Vec<(HTLCPreviousHopData, OutboundHop)>,
 		> = new_hash_map();
-		let prune_forwarded_htlc = |already_forwarded_htlcs: &mut HashMap<
-			(ChannelId, PaymentHash),
-			Vec<(HTLCPreviousHopData, OutboundHop)>,
-		>,
-		                            prev_hop: &HTLCPreviousHopData,
-		                            payment_hash: &PaymentHash| {
-			if let hash_map::Entry::Occupied(mut entry) =
-				already_forwarded_htlcs.entry((prev_hop.channel_id, *payment_hash))
-			{
-				entry.get_mut().retain(|(htlc, _)| prev_hop.htlc_id != htlc.htlc_id);
-				if entry.get().is_empty() {
-					entry.remove();
-				}
-			}
-		};
 		{
 			// If we're tracking pending payments, ensure we haven't lost any by looking at the
 			// ChannelMonitor data for any channels for which we do not have authorative state
@@ -19381,65 +19366,19 @@ impl<
 						let htlc_id = SentHTLCId::from_source(&htlc_source);
 						match htlc_source {
 							HTLCSource::PreviousHopData(prev_hop_data) => {
-								let pending_forward_matches_htlc = |info: &PendingAddHTLCInfo| {
-									info.prev_funding_outpoint == prev_hop_data.outpoint
-										&& info.prev_htlc_id == prev_hop_data.htlc_id
-								};
-
-								// If `reconstruct_manager_from_monitors` is set, we always add all inbound committed
-								// HTLCs to `decode_update_add_htlcs` in the above loop, but we need to prune from
-								// those added HTLCs if they were already forwarded to the outbound edge. Otherwise,
-								// we'll double-forward.
-								if reconstruct_manager_from_monitors {
-									dedup_decode_update_add_htlcs(
-										&mut decode_update_add_htlcs,
-										&prev_hop_data,
-										"HTLC already forwarded to the outbound edge",
-										&&logger,
-									);
-									prune_forwarded_htlc(
-										&mut already_forwarded_htlcs,
-										&prev_hop_data,
-										&htlc.payment_hash,
-									);
-								}
-
-								// The ChannelMonitor is now responsible for this HTLC's
-								// failure/success and will let us know what its outcome is. If we
-								// still have an entry for this HTLC in `forward_htlcs_legacy`,
-								// `pending_intercepted_htlcs_legacy`, or
-								// `decode_update_add_htlcs_legacy`, we were apparently not persisted
-								// after the monitor was when forwarding the payment.
-								dedup_decode_update_add_htlcs(
+								reconcile_pending_htlcs_with_monitor(
+									reconstruct_manager_from_monitors,
+									&mut already_forwarded_htlcs,
+									&mut forward_htlcs_legacy,
+									&mut pending_events_read,
+									&mut pending_intercepted_htlcs_legacy,
+									&mut decode_update_add_htlcs,
 									&mut decode_update_add_htlcs_legacy,
-									&prev_hop_data,
-									"HTLC was forwarded to the closed channel",
-									&&logger,
+									prev_hop_data,
+									&logger,
+									htlc.payment_hash,
+									monitor.channel_id(),
 								);
-								forward_htlcs_legacy.retain(|_, forwards| {
-								forwards.retain(|forward| {
-									if let HTLCForwardInfo::AddHTLC(htlc_info) = forward {
-										if pending_forward_matches_htlc(&htlc_info) {
-											log_info!(logger, "Removing pending to-forward HTLC with hash {} as it was forwarded to the closed channel {}",
-												&htlc.payment_hash, &monitor.channel_id());
-											false
-										} else { true }
-									} else { true }
-								});
-								!forwards.is_empty()
-							});
-								pending_intercepted_htlcs_legacy.retain(|intercepted_id, htlc_info| {
-								if pending_forward_matches_htlc(&htlc_info) {
-									log_info!(logger, "Removing pending intercepted HTLC with hash {} as it was forwarded to the closed channel {}",
-										&htlc.payment_hash, &monitor.channel_id());
-									pending_events_read.retain(|(event, _)| {
-										if let Event::HTLCIntercepted { intercept_id: ev_id, .. } = event {
-											intercepted_id != ev_id
-										} else { true }
-									});
-									false
-								} else { true }
-							});
 							},
 							HTLCSource::TrampolineForward { .. } => todo!(),
 							HTLCSource::OutboundRoute {
@@ -20341,6 +20280,94 @@ impl<
 	}
 }
 
+fn prune_forwarded_htlc(
+	already_forwarded_htlcs: &mut HashMap<
+		(ChannelId, PaymentHash),
+		Vec<(HTLCPreviousHopData, OutboundHop)>,
+	>,
+	prev_hop: &HTLCPreviousHopData, payment_hash: &PaymentHash,
+) {
+	if let hash_map::Entry::Occupied(mut entry) =
+		already_forwarded_htlcs.entry((prev_hop.channel_id, *payment_hash))
+	{
+		entry.get_mut().retain(|(htlc, _)| prev_hop.htlc_id != htlc.htlc_id);
+		if entry.get().is_empty() {
+			entry.remove();
+		}
+	}
+}
+
+/// Removes pending HTLC entries that the ChannelMonitor has already taken responsibility for,
+/// cleaning up state mismatches that can occur during restart.
+fn reconcile_pending_htlcs_with_monitor(
+	reconstruct_manager_from_monitors: bool,
+	already_forwarded_htlcs: &mut HashMap<
+		(ChannelId, PaymentHash),
+		Vec<(HTLCPreviousHopData, OutboundHop)>,
+	>,
+	forward_htlcs_legacy: &mut HashMap<u64, Vec<HTLCForwardInfo>>,
+	pending_events_read: &mut VecDeque<(Event, Option<EventCompletionAction>)>,
+	pending_intercepted_htlcs_legacy: &mut HashMap<InterceptId, PendingAddHTLCInfo>,
+	decode_update_add_htlcs: &mut HashMap<u64, Vec<msgs::UpdateAddHTLC>>,
+	decode_update_add_htlcs_legacy: &mut HashMap<u64, Vec<msgs::UpdateAddHTLC>>,
+	prev_hop_data: HTLCPreviousHopData, logger: &impl Logger, payment_hash: PaymentHash,
+	channel_id: ChannelId,
+) {
+	let pending_forward_matches_htlc = |info: &PendingAddHTLCInfo| {
+		info.prev_funding_outpoint == prev_hop_data.outpoint
+			&& info.prev_htlc_id == prev_hop_data.htlc_id
+	};
+
+	// If `reconstruct_manager_from_monitors` is set, we always add all inbound committed
+	// HTLCs to `decode_update_add_htlcs` in the above loop, but we need to prune from
+	// those added HTLCs if they were already forwarded to the outbound edge. Otherwise,
+	// we'll double-forward.
+	if reconstruct_manager_from_monitors {
+		dedup_decode_update_add_htlcs(
+			decode_update_add_htlcs,
+			&prev_hop_data,
+			"HTLC already forwarded to the outbound edge",
+			&&logger,
+		);
+		prune_forwarded_htlc(already_forwarded_htlcs, &prev_hop_data, &payment_hash);
+	}
+
+	// The ChannelMonitor is now responsible for this HTLC's failure/success and will let us know
+	// what its outcome is. If we still have an entry for this HTLC in `forward_htlcs_legacy`,
+	// `pending_intercepted_htlcs_legacy`, or `decode_update_add_htlcs_legacy`, we were apparently
+	// not persisted after the monitor was when forwarding the payment.
+	dedup_decode_update_add_htlcs(
+		decode_update_add_htlcs_legacy,
+		&prev_hop_data,
+		"HTLC was forwarded to the closed channel",
+		&&logger,
+	);
+	forward_htlcs_legacy.retain(|_, forwards| {
+		forwards.retain(|forward| {
+			if let HTLCForwardInfo::AddHTLC(htlc_info) = forward {
+				if pending_forward_matches_htlc(&htlc_info) {
+					log_info!(logger, "Removing pending to-forward HTLC with hash {} as it was forwarded to the closed channel {}",
+						&payment_hash, channel_id);
+					false
+				} else { true }
+			} else { true }
+		});
+		!forwards.is_empty()
+	});
+	pending_intercepted_htlcs_legacy.retain(|intercepted_id, htlc_info| {
+		if pending_forward_matches_htlc(&htlc_info) {
+			log_info!(logger, "Removing pending intercepted HTLC with hash {} as it was forwarded to the closed channel {}",
+				payment_hash, channel_id);
+			pending_events_read.retain(|(event, _)| {
+				if let Event::HTLCIntercepted { intercept_id: ev_id, .. } = event {
+					intercepted_id != ev_id
+				} else { true }
+			});
+			false
+		} else { true }
+	});
+}
+
 #[cfg(test)]
 mod tests {
 	use crate::events::{ClosureReason, Event, HTLCHandlingFailureType};