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

@@ -5737,6 +5737,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.
 	///
@@ -5757,22 +5817,34 @@ 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();
+					#[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;
 				}
-				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 {
@@ -5869,22 +5941,34 @@ 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();
+						#[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;
 					}
-					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