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

@@ -5678,6 +5678,64 @@ where
 		self.pending_outbound_payments.clear_pending_payments()
 	}
 
+	/// (Partially) handles an [`EventCompletionAction`]. If `true` is returned the same `action`
+	/// should be processed again.
+	fn handle_post_event_action(&self, action: &EventCompletionAction) -> bool {
+		match action {
+			EventCompletionAction::ReleaseRAAChannelMonitorUpdate {
+				channel_funding_outpoint, counterparty_node_id
+			} => {
+				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.as_ref() == Some(&action))
+					{
+						// 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 false;
+					}
+					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 {
+								return true;
+							}
+						} 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));
+				}
+			},
+		}
+		false
+	}
+
+	fn handle_post_event_actions(&self, actions: Vec<EventCompletionAction>) {
+		for action in actions {
+			loop {
+				if !self.handle_post_event_action(&action) { break; }
+			}
+		}
+	}
+
 	/// Processes any events asynchronously in the order they were generated since the last call
 	/// using the given event handler.
 	///
@@ -5698,22 +5756,32 @@ where
 		}
 
 		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();
+					}
+					handler(event).await;
+					let mut pending_events = self.pending_events.lock().unwrap();
+					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;
 				}
-				handler(event).await;
-				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 {
@@ -5810,22 +5878,32 @@ where
 			}
 
 			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();
+						}
+						handler.handle_event(event);
+						let mut pending_events = self.pending_events.lock().unwrap();
+						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;
 					}
-					handler.handle_event(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
 			}
 
 			result