Gather to-decode HTLC fwds from channels on manager read

We have an overarching goal of (mostly) getting rid of ChannelManager
persistence and rebuilding the ChannelManager's state from existing
ChannelMonitors, due to issues when the two structs are out-of-sync on restart.
The main issue that can arise is channel force closure.

Here we start this process by rebuilding
ChannelManager::decode_update_add_htlcs from the Channels, which will soon be
included in the ChannelMonitors as part of a different series of PRs.

The newly built map is not yet used but will be in the next commit.

Author: valentinewallace

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(

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,35 @@ where
 			}
 		}
 
+		// In the future, the set of pending HTLCs will be pulled from `Channel{Monitor}` data and
+		// placed in `ChannelManager::decode_update_add_htlcs` on read, to be handled on the next call
+		// to `process_pending_htlc_forwards`. This is part of a larger effort to remove the requirement
+		// of regularly persisting the `ChannelManager`. However, this new pipeline cannot currently
+		// handle inbound HTLC receives, some of which may be present in `failed_htlcs`, so continue
+		// handling HTLCs present in `failed_htlcs` during deserialization for now.
+		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,