Fix replay of preimage monitor events for fwds

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.

If the outbound edge of a forwarded payment is resolved on-chain via preimage,
and we get a monitor event for it, we want to mark that monitor event resolved
once the preimage is durably persisted in the inbound edge. Hence, we add the
monitor event resolution to the completion of the inbound edge's preimage
monitor update.

Author: valentinewallace

lightning/src/ln/channelmanager.rs

@@ -1475,6 +1475,9 @@ pub(crate) enum MonitorUpdateCompletionAction {
 	EmitEventOptionAndFreeOtherChannel {
 		event: Option<events::Event>,
 		downstream_counterparty_and_funding_outpoint: EventUnblockedChannel,
+		/// If this action originated from a [`MonitorEvent`], the source to acknowledge once the
+		/// action is handled.
+		monitor_event_source: Option<MonitorEventSource>,
 	},
 	/// Indicates we should immediately resume the operation of another channel, unless there is
 	/// some other reason why the channel is blocked. In practice this simply means immediately
@@ -1492,6 +1495,9 @@ pub(crate) enum MonitorUpdateCompletionAction {
 		downstream_counterparty_node_id: PublicKey,
 		blocking_action: RAAMonitorUpdateBlockingAction,
 		downstream_channel_id: ChannelId,
+		/// If this action originated from a [`MonitorEvent`], the source to acknowledge once the
+		/// action is handled.
+		monitor_event_source: Option<MonitorEventSource>,
 	},
 	/// Indicates that one or more [`MonitorEvent`]s should be acknowledged once the associated
 	/// [`ChannelMonitorUpdate`] has been durably persisted.
@@ -1509,13 +1515,15 @@ impl_writeable_tlv_based_enum_upgradable!(MonitorUpdateCompletionAction,
 		(0, downstream_counterparty_node_id, required),
 		(4, blocking_action, upgradable_required),
 		(5, downstream_channel_id, required),
+		(7, monitor_event_source, option),
 	},
 	(2, EmitEventOptionAndFreeOtherChannel) => {
 		// LDK prior to 0.3 required this field. It will not be present for trampoline payments
 		// with multiple incoming HTLCS, so nodes cannot downgrade while trampoline payments
 		// are in the process of being resolved.
 		(0, event, upgradable_option),
 		(1, downstream_counterparty_and_funding_outpoint, upgradable_required),
+		(3, monitor_event_source, option),
 	},
 	(3, AckMonitorEvents) => {
 		(0, monitor_events_to_ack, required_vec),
@@ -9438,6 +9446,7 @@ impl<
 		startup_replay: bool, next_channel_counterparty_node_id: PublicKey,
 		next_channel_outpoint: OutPoint, next_channel_id: ChannelId, hop_data: HTLCPreviousHopData,
 		attribution_data: Option<AttributionData>, send_timestamp: Option<Duration>,
+		monitor_event_source: Option<MonitorEventSource>,
 	) {
 		let _prev_channel_id = hop_data.channel_id;
 		let completed_blocker = RAAMonitorUpdateBlockingAction::from_prev_hop_data(&hop_data);
@@ -9527,6 +9536,7 @@ impl<
 							downstream_counterparty_node_id: chan_to_release.counterparty_node_id,
 							downstream_channel_id: chan_to_release.channel_id,
 							blocking_action: chan_to_release.blocking_action,
+							monitor_event_source,
 						}),
 						None,
 					)
@@ -9542,6 +9552,7 @@ impl<
 						Some(MonitorUpdateCompletionAction::EmitEventOptionAndFreeOtherChannel {
 							event,
 							downstream_counterparty_and_funding_outpoint: chan_to_release,
+							monitor_event_source,
 						}),
 						None,
 					)
@@ -9726,8 +9737,12 @@ impl<
 								downstream_counterparty_node_id: node_id,
 								blocking_action: blocker,
 								downstream_channel_id: channel_id,
+								monitor_event_source,
 							} = action
 							{
+								if let Some(source) = monitor_event_source {
+									self.chain_monitor.ack_monitor_event(source);
+								}
 								if let Some(peer_state_mtx) = per_peer_state.get(&node_id) {
 									let mut peer_state = peer_state_mtx.lock().unwrap();
 									if let Some(blockers) = peer_state
@@ -9986,6 +10001,8 @@ This indicates a bug inside LDK. Please report this error at https://github.com/
 					hop_data,
 					attribution_data,
 					send_timestamp,
+					monitor_event_id
+						.map(|id| MonitorEventSource { event_id: id, channel_id: next_channel_id }),
 				);
 			},
 			HTLCSource::TrampolineForward { previous_hop_data, .. } => {
@@ -10029,6 +10046,19 @@ This indicates a bug inside LDK. Please report this error at https://github.com/
 						current_previous_hop_data,
 						attribution_data.clone(),
 						send_timestamp,
+						// Only pass the monitor event ID for the first hop. Once one
+						// inbound channel durably has the preimage, the outbound
+						// monitor event can be acked. The preimage remains in the
+						// outbound monitor's storage and will be replayed to all
+						// remaining inbound hops on restart via pending_claims_to_replay.
+						if i == 0 {
+							monitor_event_id.map(|id| MonitorEventSource {
+								event_id: id,
+								channel_id: next_channel_id,
+							})
+						} else {
+							None
+						},
 					);
 				}
 			},
@@ -10255,7 +10285,11 @@ This indicates a bug inside LDK. Please report this error at https://github.com/
 				MonitorUpdateCompletionAction::EmitEventOptionAndFreeOtherChannel {
 					event,
 					downstream_counterparty_and_funding_outpoint,
+					monitor_event_source,
 				} => {
+					if let Some(source) = monitor_event_source {
+						self.chain_monitor.ack_monitor_event(source);
+					}
 					if let Some(event) = event {
 						self.pending_events.lock().unwrap().push_back((event, None));
 					}
@@ -10269,7 +10303,11 @@ This indicates a bug inside LDK. Please report this error at https://github.com/
 					downstream_counterparty_node_id,
 					downstream_channel_id,
 					blocking_action,
+					monitor_event_source,
 				} => {
+					if let Some(source) = monitor_event_source {
+						self.chain_monitor.ack_monitor_event(source);
+					}
 					self.handle_monitor_update_release(
 						downstream_counterparty_node_id,
 						downstream_channel_id,