Gather to-decode HTLC fwds from channels on manager read

diff --git a/lightning/src/ln/channel.rs b/lightning/src/ln/channel.rs
index ed6f6ce..cb45540 100644
--- a/lightning/src/ln/channel.rs
+++ b/lightning/src/ln/channel.rs
@@ -7778,6 +7778,20 @@ where
 		Ok(())
 	}
 
+	/// Useful for reconstructing the set of pending HTLCs when deserializing the `ChannelManager`.
+	pub(super) fn get_inbound_committed_update_adds(&self) -> Vec<msgs::UpdateAddHTLC> {
+		self.context
+			.pending_inbound_htlcs
+			.iter()
+			.filter_map(|htlc| match htlc.state {
+				InboundHTLCState::Committed { ref update_add_htlc_opt } => {
+					update_add_htlc_opt.clone()
+				},
+				_ => None,
+			})
+			.collect()
+	}
+
 	/// Marks an outbound HTLC which we have received update_fail/fulfill/malformed
 	#[inline]
 	fn mark_outbound_htlc_removed(
diff --git a/lightning/src/ln/channelmanager.rs b/lightning/src/ln/channelmanager.rs
index a854bb7..080ecef 100644
--- a/lightning/src/ln/channelmanager.rs
+++ b/lightning/src/ln/channelmanager.rs
@@ -17358,6 +17358,7 @@ where
 			decode_update_add_htlcs_legacy.unwrap_or_else(|| new_hash_map());
 		let mut pending_intercepted_htlcs_legacy =
 			pending_intercepted_htlcs_legacy.unwrap_or_else(|| new_hash_map());
+		let mut decode_update_add_htlcs = new_hash_map();
 		let peer_storage_dir: Vec<(PublicKey, Vec<u8>)> = peer_storage_dir.unwrap_or_else(Vec::new);
 		if fake_scid_rand_bytes.is_none() {
 			fake_scid_rand_bytes = Some(args.entropy_source.get_secure_random_bytes());
@@ -17669,6 +17670,21 @@ where
 					let mut peer_state_lock = peer_state_mtx.lock().unwrap();
 					let peer_state = &mut *peer_state_lock;
 					is_channel_closed = !peer_state.channel_by_id.contains_key(channel_id);
+					if let Some(chan) = peer_state.channel_by_id.get(channel_id) {
+						if let Some(funded_chan) = chan.as_funded() {
+							let inbound_committed_update_adds =
+								funded_chan.get_inbound_committed_update_adds();
+							if !inbound_committed_update_adds.is_empty() {
+								// Reconstruct `ChannelManager::decode_update_add_htlcs` from the serialized
+								// `Channel`, as part of removing the requirement to regularly persist the
+								// `ChannelManager`.
+								decode_update_add_htlcs.insert(
+									funded_chan.context.outbound_scid_alias(),
+									inbound_committed_update_adds,
+								);
+							}
+						}
+					}
 				}
 
 				if is_channel_closed {
@@ -17727,9 +17743,15 @@ where
 								};
 								// 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` or
-								// `pending_intercepted_htlcs`, we were apparently not persisted after
-								// the monitor was when forwarding the payment.
+								// still have an entry for this HTLC in `forward_htlcs`,
+								// `pending_intercepted_htlcs`, or `decode_update_add_htlcs`, we were apparently not
+								// persisted after the monitor was when forwarding the payment.
+								dedup_decode_update_add_htlcs(
+									&mut decode_update_add_htlcs,
+									&prev_hop_data,
+									"HTLC was forwarded to the closed channel",
+									&args.logger,
+								);
 								dedup_decode_update_add_htlcs(
 									&mut decode_update_add_htlcs_legacy,
 									&prev_hop_data,
@@ -18220,6 +18242,31 @@ where
 			}
 		}
 
+		// De-duplicate HTLCs that are present in both `failed_htlcs` and `decode_update_add_htlcs`.
+		// Omitting this de-duplication could lead to redundant HTLC processing and/or bugs.
+		for (src, _, _, _, _, _) in failed_htlcs.iter() {
+			if let HTLCSource::PreviousHopData(prev_hop_data) = src {
+				dedup_decode_update_add_htlcs(
+					&mut decode_update_add_htlcs,
+					prev_hop_data,
+					"HTLC was failed backwards during manager read",
+					&args.logger,
+				);
+			}
+		}
+
+		// See above comment on `failed_htlcs`.
+		for htlcs in claimable_payments.values().map(|pmt| &pmt.htlcs) {
+			for prev_hop_data in htlcs.iter().map(|h| &h.prev_hop) {
+				dedup_decode_update_add_htlcs(
+					&mut decode_update_add_htlcs,
+					prev_hop_data,
+					"HTLC was already decoded and marked as a claimable payment",
+					&args.logger,
+				);
+			}
+		}
+
 		let best_block = BestBlock::new(best_block_hash, best_block_height);
 		let flow = OffersMessageFlow::new(
 			chain_hash,