Handle EventCompletionActions after events complete

This adds handling of the new `EventCompletionAction`s after
`Event`s are handled, letting `ChannelMonitorUpdate`s which were
blocked fly after a relevant `Event`.

Author: TheBlueMatt

lightning/src/ln/channelmanager.rs

@@ -1706,23 +1706,39 @@ macro_rules! process_events_body {
 		}
 
 		let _single_processor = $self.pending_events_processor.lock().unwrap();
-		let mut next_event = $self.pending_events.lock().unwrap().front().map(|ev| (*ev).clone());
 		loop {
-			if let Some((event, _action)) = next_event {
-				result = NotifyOption::DoPersist;
-				let ev_clone;
-				#[cfg(debug_assertions)] {
-					ev_clone = event.clone();
+			let mut next_event = $self.pending_events.lock().unwrap().front().map(|ev| (*ev).clone());
+			let mut post_event_actions = Vec::new();
+			loop {
+				if let Some((event, action_opt)) = next_event {
+					result = NotifyOption::DoPersist;
+					let _ev_clone: Event;
+					#[cfg(debug_assertions)] {
+						_ev_clone = event.clone();
+					}
+					$event_to_handle = event;
+					$handle_event;
+					let mut pending_events = $self.pending_events.lock().unwrap();
+					// We're required to take the `pending_events_processor` lock any time we
+					// change `pending_events`, however have no general way to enforce that.
+					// Instead, here, we assert that the next event hasn't changed out from under
+					// us.
+					#[cfg(debug_assertions)] {
+						debug_assert_eq!(_ev_clone, pending_events.front().unwrap().0);
+					}
+					debug_assert_eq!(action_opt, pending_events.front().unwrap().1);
+					if let Some(action) = action_opt {
+						post_event_actions.push(action);
+					}
+					pending_events.pop_front();
+					next_event = pending_events.front().map(|ev| ev.clone());
+				} else {
+					break;
 				}
-				$event_to_handle = event;
-				$handle_event;
-				let mut pending_events = $self.pending_events.lock().unwrap();
-				debug_assert_eq!(ev_clone, pending_events.front().unwrap().0);
-				pending_events.pop_front();
-				next_event = pending_events.front().map(|ev| ev.clone());
-			} else {
-				break;
 			}
+			if post_event_actions.is_empty() { break; }
+			$self.handle_post_event_actions(post_event_actions);
+			// If we had some actions, go around again as we may have more events now
 		}
 
 		if result == NotifyOption::DoPersist {
@@ -5860,6 +5876,66 @@ where
 		self.pending_outbound_payments.clear_pending_payments()
 	}
 
+	fn handle_monitor_update_release(&self, counterparty_node_id: PublicKey, channel_funding_outpoint: OutPoint) {
+		loop {
+			let per_peer_state = self.per_peer_state.read().unwrap();
+			if let Some(peer_state_mtx) = per_peer_state.get(&counterparty_node_id) {
+				let mut peer_state_lck = peer_state_mtx.lock().unwrap();
+				let peer_state = &mut *peer_state_lck;
+				if self.pending_events.lock().unwrap().iter()
+					.any(|(_ev, action_opt)| action_opt == &Some(EventCompletionAction::ReleaseRAAChannelMonitorUpdate {
+						channel_funding_outpoint, counterparty_node_id
+					}))
+				{
+					// Check that, while holding the peer lock, we don't have another event
+					// blocking any monitor updates for this channel. If we do, let those
+					// events be the ones that ultimately release the monitor update(s).
+					log_trace!(self.logger, "Delaying monitor unlock for channel {} as another event is pending",
+						log_bytes!(&channel_funding_outpoint.to_channel_id()[..]));
+					return;
+				}
+				if let hash_map::Entry::Occupied(mut chan) = peer_state.channel_by_id.entry(channel_funding_outpoint.to_channel_id()) {
+					debug_assert_eq!(chan.get().get_funding_txo().unwrap(), channel_funding_outpoint);
+					if let Some((monitor_update, further_update_exists)) = chan.get_mut().fly_next_unflown_monitor_update() {
+						log_debug!(self.logger, "Unlocking monitor updating for channel {} and updating monitor",
+							log_bytes!(&channel_funding_outpoint.to_channel_id()[..]));
+						let update_res = self.chain_monitor.update_channel(channel_funding_outpoint, monitor_update);
+						let update_id = monitor_update.update_id;
+						let _ = handle_error!(self,
+							handle_new_monitor_update!(self, update_res, update_id,
+								peer_state_lck, peer_state, per_peer_state, chan),
+							counterparty_node_id);
+						if further_update_exists {
+							// If there are more `ChannelMonitorUpdate`s to process, restart at the
+							// top of the loop.
+							continue;
+						}
+					} else {
+						log_trace!(self.logger, "Unlocked monitor updating for channel {} without monitors to update",
+							log_bytes!(&channel_funding_outpoint.to_channel_id()[..]));
+					}
+				}
+			} else {
+				log_debug!(self.logger,
+					"Got a release post-RAA monitor update for peer {} but the channel is gone",
+					log_pubkey!(counterparty_node_id));
+			}
+			break;
+		}
+	}
+
+	fn handle_post_event_actions(&self, actions: Vec<EventCompletionAction>) {
+		for action in actions {
+			match action {
+				EventCompletionAction::ReleaseRAAChannelMonitorUpdate {
+					channel_funding_outpoint, counterparty_node_id
+				} => {
+					self.handle_monitor_update_release(counterparty_node_id, channel_funding_outpoint);
+				}
+			}
+		}
+	}
+
 	/// Processes any events asynchronously in the order they were generated since the last call
 	/// using the given event handler.
 	///