Ack monitor events on revoke_and_ack

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 build on recent commits by ACK'ing monitor events for forward failures
once the monitor update that marks them as failed on the inbound edge is
complete, when the HTLC was irrevocably failed via revoke_and_ack.

Author: valentinewallace

lightning/src/ln/channel.rs

@@ -8587,7 +8587,7 @@ where
 		(
 			Vec<(HTLCSource, PaymentHash)>,
 			Vec<(StaticInvoice, BlindedMessagePath)>,
-			Option<ChannelMonitorUpdate>,
+			Option<(ChannelMonitorUpdate, Vec<MonitorEventSource>)>,
 		),
 		ChannelError,
 	> {
@@ -8898,6 +8898,7 @@ where
 		} else {
 			"Blocked"
 		};
+		let mut monitor_events_to_ack = Vec::new();
 		macro_rules! return_with_htlcs_to_fail {
 			($htlcs_to_fail: expr) => {
 				if !release_monitor {
@@ -8906,18 +8907,21 @@ where
 						.push(PendingChannelMonitorUpdate { update: monitor_update });
 					return Ok(($htlcs_to_fail, static_invoices, None));
 				} else {
-					return Ok(($htlcs_to_fail, static_invoices, Some(monitor_update)));
+					let events_to_ack = core::mem::take(&mut monitor_events_to_ack);
+					return Ok((
+						$htlcs_to_fail,
+						static_invoices,
+						Some((monitor_update, events_to_ack)),
+					));
 				}
 			};
 		}
 
 		self.context.monitor_pending_update_adds.append(&mut pending_update_adds);
 
 		match self.maybe_free_holding_cell_htlcs(fee_estimator, logger) {
-			// TODO: Thread monitor_events_to_ack through the revoke_and_ack return
-			// value so the ChannelManager can attach an AckMonitorEvents completion
-			// action to this monitor update.
-			(Some((mut additional_update, _monitor_events_to_ack)), htlcs_to_fail) => {
+			(Some((mut additional_update, holding_cell_monitor_events_to_ack)), htlcs_to_fail) => {
+				monitor_events_to_ack = holding_cell_monitor_events_to_ack;
 				// free_holding_cell_htlcs may bump latest_monitor_id multiple times but we want them to be
 				// strictly increasing by one, so decrement it here.
 				self.context.latest_monitor_update_id = monitor_update.update_id;

lightning/src/ln/channelmanager.rs

@@ -12981,7 +12981,16 @@ This indicates a bug inside LDK. Please report this error at https://github.com/
 							*counterparty_node_id);
 						let (htlcs_to_fail, static_invoices, monitor_update_opt) = try_channel_entry!(self, peer_state,
 							chan.revoke_and_ack(&msg, &self.fee_estimator, &&logger, mon_update_blocked), chan_entry);
-						if let Some(monitor_update) = monitor_update_opt {
+						if let Some((monitor_update, monitor_events_to_ack)) = monitor_update_opt {
+							if !monitor_events_to_ack.is_empty() {
+								peer_state
+									.monitor_update_blocked_actions
+									.entry(msg.channel_id)
+									.or_default()
+									.push(MonitorUpdateCompletionAction::AckMonitorEvents {
+										monitor_events_to_ack,
+									});
+							}
 							let funding_txo = funding_txo_opt
 								.expect("Funding outpoint must have been set for RAA handling to succeed");
 							if let Some(data) = self.handle_new_monitor_update(