Ack monitor events when failing-to-fail-backwards

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
if we go to fail on the inbound edge and get an error when queueing the
backwards failure. If we get an error in this case it means the failure is
duplicate or the channel is closed, and in either case it's fine to mark the
event as resolved.

Author: valentinewallace

lightning/src/ln/channelmanager.rs

@@ -8196,6 +8196,7 @@ impl<
 								&&logger,
 							),
 							htlc_id,
+							monitor_event_source,
 						))
 					} else {
 						self.forwarding_channel_not_found(
@@ -8228,7 +8229,7 @@ impl<
 							monitor_event_source,
 							&&logger,
 						);
-						Some((res, htlc_id))
+						Some((res, htlc_id, monitor_event_source))
 					} else {
 						self.forwarding_channel_not_found(
 							core::iter::once(forward_info).chain(draining_pending_forwards),
@@ -8241,8 +8242,12 @@ impl<
 					}
 				},
 			};
-			if let Some((queue_fail_htlc_res, htlc_id)) = queue_fail_htlc_res {
+			if let Some((queue_fail_htlc_res, htlc_id, monitor_event_source)) = queue_fail_htlc_res
+			{
 				if let Err(e) = queue_fail_htlc_res {
+					if let Some(source) = monitor_event_source {
+						self.chain_monitor.ack_monitor_event(source);
+					}
 					if let ChannelError::Ignore(msg) = e {
 						if let Some(chan) = peer_state
 							.channel_by_id