Persistent HTLC forward claim monitor events

Currently, the resolution of HTLCs (and decisions on when HTLCs can be
forwarded) is the responsibility of Channel objects (a part of ChannelManager)
until the channel is closed, and then the ChannelMonitor thereafter. This leads
to some complexity around race conditions for HTLCs right around channel
closure. Additionally, there is lots of complexity reconstructing the state of
all HTLCs in the ChannelManager deserialization/loading logic.

Instead, we want to do all resolution in ChannelMonitors (in response to
ChannelMonitorUpdates) and pass them back to ChannelManager in the form of
MonitorEvents (similar to how HTLCs are resolved after channels are closed). In
order to have reliable resolution, we'll need to keep MonitorEvents around in
the ChannelMonitor until the ChannelManager has finished processing them. This
will simplify things - on restart instead of examining the set of HTLCs in
monitors we can simply replay all the pending MonitorEvents.

Here we add MonitorEvent resolution of forwarded HTLC claims, which allows
us to remove usage of RAA monitor update blocking actions and a significant
amount of complex startup reconstruction code.

Author: valentinewallace

lightning/src/chain/channelmonitor.rs

@@ -2289,6 +2289,18 @@ impl<Signer: EcdsaChannelSigner> ChannelMonitor<Signer> {
 		inner.ack_monitor_event(event_id);
 	}
 
+	/// Returns true if this monitor has any pending or provided `HTLCEvent`s containing a
+	/// payment preimage, indicating a claim is in progress.
+	#[cfg(any(test, feature = "_test_utils"))]
+	pub fn has_pending_claim_monitor_events(&self) -> bool {
+		let inner = self.inner.lock().unwrap();
+		inner.pending_monitor_events.iter().chain(inner.provided_monitor_events.iter()).any(
+			|(_, ev)| {
+				matches!(ev, MonitorEvent::HTLCEvent(HTLCUpdate { payment_preimage: Some(_), .. }))
+			},
+		)
+	}
+
 	/// Enables persistent monitor events mode. When enabled, monitor events are retained until
 	/// explicitly acked rather than cleared on read.
 	pub(crate) fn set_persistent_events_enabled(&self, enabled: bool) {
@@ -3888,20 +3900,32 @@ impl<Signer: EcdsaChannelSigner> ChannelMonitorImpl<Signer> {
 		self.prev_holder_htlc_data = Some(htlc_data);
 
 		for claim in claimed_htlcs {
-			#[cfg(debug_assertions)]
-			{
-				let cur_counterparty_htlcs = self
-					.funding
-					.counterparty_claimable_outpoints
-					.get(&self.funding.current_counterparty_commitment_txid.unwrap())
-					.unwrap();
-				assert!(cur_counterparty_htlcs.iter().any(|(_, source_opt)| {
+			let htlc_opt = self
+				.funding
+				.counterparty_claimable_outpoints
+				.get(&self.funding.current_counterparty_commitment_txid.unwrap())
+				.unwrap()
+				.iter()
+				.find_map(|(htlc, source_opt)| {
 					if let Some(source) = source_opt {
-						SentHTLCId::from_source(source) == claim.htlc_id
-					} else {
-						false
+						if SentHTLCId::from_source(source) == claim.htlc_id {
+							return Some((htlc, source));
+						}
 					}
-				}));
+					None
+				});
+			debug_assert!(htlc_opt.is_some());
+			if self.persistent_events_enabled {
+				if let Some((htlc, source)) = htlc_opt {
+					self.push_monitor_event(MonitorEvent::HTLCEvent(HTLCUpdate {
+						payment_hash: htlc.payment_hash,
+						payment_preimage: Some(claim.preimage),
+						source: *source.clone(),
+						htlc_value_satoshis: Some(htlc.amount_msat),
+						skimmed_fee_msat: claim.skimmed_fee_msat,
+						next_user_channel_id: self.user_channel_id,
+					}));
+				}
 			}
 			self.counterparty_fulfilled_htlcs.insert(claim.htlc_id, claim.preimage);
 		}