Fix replay of monitor events for outbounds

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 an outbound payment resolved on-chain and we get a monitor event for it, we
want to mark that monitor event as resolved once the user has processed a
PaymentSent or a PaymentFailed event.

Author: valentinewallace

lightning/src/ln/channelmanager.rs

@@ -1544,19 +1544,34 @@ enum PostMonitorUpdateChanResume {
 	},
 }
 
-#[derive(Clone, Debug, PartialEq, Eq)]
+#[derive(Clone, Debug, Eq)]
 pub(crate) struct PaymentCompleteUpdate {
 	counterparty_node_id: PublicKey,
 	channel_funding_outpoint: OutPoint,
 	channel_id: ChannelId,
 	htlc_id: SentHTLCId,
+	/// If this payment resolution originated from a [`MonitorEvent`], the event ID to acknowledge
+	/// once the completion action has been durably handled.
+	monitor_event_id: Option<u64>,
+}
+
+impl PartialEq for PaymentCompleteUpdate {
+	fn eq(&self, other: &Self) -> bool {
+		self.counterparty_node_id == other.counterparty_node_id
+			&& self.channel_funding_outpoint == other.channel_funding_outpoint
+			&& self.channel_id == other.channel_id
+			&& self.htlc_id == other.htlc_id
+		// monitor_event_id is excluded: it's metadata about the event source,
+		// not part of the semantic identity of the payment resolution.
+	}
 }
 
 impl_writeable_tlv_based!(PaymentCompleteUpdate, {
 	(1, channel_funding_outpoint, required),
 	(3, counterparty_node_id, required),
 	(5, channel_id, required),
 	(7, htlc_id, required),
+	(9, monitor_event_id, option),
 });
 
 #[derive(Clone, Debug, PartialEq, Eq)]
@@ -9871,6 +9886,7 @@ This indicates a bug inside LDK. Please report this error at https://github.com/
 		next_channel_counterparty_node_id: PublicKey, next_channel_outpoint: OutPoint,
 		next_channel_id: ChannelId, next_user_channel_id: Option<u128>,
 		attribution_data: Option<AttributionData>, send_timestamp: Option<Duration>,
+		monitor_event_id: Option<u64>,
 	) {
 		let startup_replay =
 			!self.background_events_processed_since_startup.load(Ordering::Acquire);
@@ -9889,6 +9905,7 @@ This indicates a bug inside LDK. Please report this error at https://github.com/
 						channel_funding_outpoint: next_channel_outpoint,
 						channel_id: next_channel_id,
 						htlc_id,
+						monitor_event_id,
 					};
 					Some(EventCompletionAction::ReleasePaymentCompleteChannelMonitorUpdate(release))
 				} else {
@@ -12474,6 +12491,7 @@ This indicates a bug inside LDK. Please report this error at https://github.com/
 			Some(next_user_channel_id),
 			msg.attribution_data,
 			send_timestamp,
+			None,
 		);
 
 		Ok(())
@@ -13297,6 +13315,7 @@ This indicates a bug inside LDK. Please report this error at https://github.com/
 								None,
 								None,
 								None,
+								Some(event_id),
 							);
 						} else {
 							log_trace!(logger, "Failing HTLC from our monitor");
@@ -13309,6 +13328,7 @@ This indicates a bug inside LDK. Please report this error at https://github.com/
 								channel_funding_outpoint: funding_outpoint,
 								channel_id,
 								htlc_id: SentHTLCId::from_source(&htlc_update.source),
+								monitor_event_id: Some(event_id),
 							});
 							self.fail_htlc_backwards_internal(
 								&htlc_update.source,
@@ -15086,8 +15106,13 @@ impl<
 						channel_funding_outpoint,
 						channel_id,
 						htlc_id,
+						monitor_event_id,
 					},
 				) => {
+					if let Some(event_id) = monitor_event_id {
+						self.chain_monitor
+							.ack_monitor_event(MonitorEventSource { channel_id, event_id });
+					}
 					let per_peer_state = self.per_peer_state.read().unwrap();
 					let mut peer_state_lock = per_peer_state
 						.get(&counterparty_node_id)
@@ -19460,6 +19485,7 @@ impl<
 										channel_funding_outpoint: monitor.get_funding_txo(),
 										channel_id: monitor.channel_id(),
 										htlc_id,
+										monitor_event_id: None,
 									};
 									let mut compl_action = Some(
 									EventCompletionAction::ReleasePaymentCompleteChannelMonitorUpdate(update)
@@ -19539,6 +19565,7 @@ impl<
 							channel_funding_outpoint: monitor.get_funding_txo(),
 							channel_id: monitor.channel_id(),
 							htlc_id: SentHTLCId::from_source(&htlc_source),
+							monitor_event_id: None,
 						});
 
 						failed_htlcs.push((
@@ -20296,6 +20323,7 @@ impl<
 				downstream_user_channel_id,
 				None,
 				None,
+				None,
 			);
 		}