Ack monitor events on check_free_peer_holding_cells

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.

Author: valentinewallace

lightning/src/ln/channel.rs

@@ -8331,7 +8331,7 @@ where
 	/// returns `(None, Vec::new())`.
 	pub fn maybe_free_holding_cell_htlcs<F: FeeEstimator, L: Logger>(
 		&mut self, fee_estimator: &LowerBoundedFeeEstimator<F>, logger: &L,
-	) -> (Option<ChannelMonitorUpdate>, Vec<(HTLCSource, PaymentHash)>) {
+	) -> (Option<(ChannelMonitorUpdate, Vec<MonitorEventSource>)>, Vec<(HTLCSource, PaymentHash)>) {
 		if matches!(self.context.channel_state, ChannelState::ChannelReady(_))
 			&& self.context.channel_state.can_generate_new_commitment()
 		{
@@ -8345,7 +8345,7 @@ where
 	/// for our counterparty.
 	fn free_holding_cell_htlcs<F: FeeEstimator, L: Logger>(
 		&mut self, fee_estimator: &LowerBoundedFeeEstimator<F>, logger: &L,
-	) -> (Option<ChannelMonitorUpdate>, Vec<(HTLCSource, PaymentHash)>) {
+	) -> (Option<(ChannelMonitorUpdate, Vec<MonitorEventSource>)>, Vec<(HTLCSource, PaymentHash)>) {
 		assert!(matches!(self.context.channel_state, ChannelState::ChannelReady(_)));
 		assert!(!self.context.channel_state.is_monitor_update_in_progress());
 		assert!(!self.context.channel_state.is_quiescent());
@@ -8375,6 +8375,7 @@ where
 			let mut update_fulfill_count = 0;
 			let mut update_fail_count = 0;
 			let mut htlcs_to_fail = Vec::new();
+			let mut monitor_events_to_ack = Vec::new();
 			for htlc_update in htlc_updates.drain(..) {
 				// Note that this *can* fail, though it should be due to rather-rare conditions on
 				// fee races with adding too many outputs which push our total payments just over
@@ -8470,31 +8471,41 @@ where
 						htlc_id,
 						ref err_packet,
 						monitor_event_source,
-					} => Some(
-						self.fail_htlc(
-							htlc_id,
-							err_packet.clone(),
-							false,
-							monitor_event_source,
-							logger,
+					} => {
+						if let Some(source) = monitor_event_source {
+							monitor_events_to_ack.push(source);
+						}
+						Some(
+							self.fail_htlc(
+								htlc_id,
+								err_packet.clone(),
+								false,
+								monitor_event_source,
+								logger,
+							)
+							.map(|fail_msg_opt| fail_msg_opt.map(|_| ())),
 						)
-						.map(|fail_msg_opt| fail_msg_opt.map(|_| ())),
-					),
+					},
 					&HTLCUpdateAwaitingACK::FailMalformedHTLC {
 						htlc_id,
 						failure_code,
 						sha256_of_onion,
 						monitor_event_source,
-					} => Some(
-						self.fail_htlc(
-							htlc_id,
-							(sha256_of_onion, failure_code),
-							false,
-							monitor_event_source,
-							logger,
+					} => {
+						if let Some(source) = monitor_event_source {
+							monitor_events_to_ack.push(source);
+						}
+						Some(
+							self.fail_htlc(
+								htlc_id,
+								(sha256_of_onion, failure_code),
+								false,
+								monitor_event_source,
+								logger,
+							)
+							.map(|fail_msg_opt| fail_msg_opt.map(|_| ())),
 						)
-						.map(|fail_msg_opt| fail_msg_opt.map(|_| ())),
-					),
+					},
 				};
 				if let Some(res) = fail_htlc_res {
 					match res {
@@ -8546,7 +8557,11 @@ where
 				Vec::new(),
 				logger,
 			);
-			(self.push_ret_blockable_mon_update(monitor_update), htlcs_to_fail)
+			(
+				self.push_ret_blockable_mon_update(monitor_update)
+					.map(|upd| (upd, monitor_events_to_ack)),
+				htlcs_to_fail,
+			)
 		} else {
 			(None, Vec::new())
 		}
@@ -8899,7 +8914,10 @@ where
 		self.context.monitor_pending_update_adds.append(&mut pending_update_adds);
 
 		match self.maybe_free_holding_cell_htlcs(fee_estimator, logger) {
-			(Some(mut additional_update), htlcs_to_fail) => {
+			// 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) => {
 				// 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

@@ -13782,7 +13782,16 @@ This indicates a bug inside LDK. Please report this error at https://github.com/
 			);
 			if monitor_opt.is_some() || !holding_cell_failed_htlcs.is_empty() {
 				let update_res = monitor_opt
-					.map(|monitor_update| {
+					.map(|(monitor_update, monitor_events_to_ack)| {
+						if !monitor_events_to_ack.is_empty() {
+							peer_state
+								.monitor_update_blocked_actions
+								.entry(*chan_id)
+								.or_default()
+								.push(MonitorUpdateCompletionAction::AckMonitorEvents {
+									monitor_events_to_ack,
+								});
+						}
 						self.handle_new_monitor_update(
 							&mut peer_state.in_flight_monitor_updates,
 							&mut peer_state.monitor_update_blocked_actions,