Handle MonitorUpdateCompletionActions in monitor update completion

In a previous PR, we added a `MonitorUpdateCompletionAction` enum
which described actions to take after a `ChannelMonitorUpdate`
persistence completes. At the time, it was only used to execute
actions in-line, however in the next commit we'll start (correctly)
leaving the existing actions until after monitor updates complete.

Author: TheBlueMatt

lightning/src/ln/channelmanager.rs

@@ -445,6 +445,7 @@ enum BackgroundEvent {
 	ClosingMonitorUpdate((OutPoint, ChannelMonitorUpdate)),
 }
 
+#[derive(Debug)]
 pub(crate) enum MonitorUpdateCompletionAction {
 	/// Indicates that a payment ultimately destined for us was claimed and we should emit an
 	/// [`events::Event::PaymentClaimed`] to the user if we haven't yet generated such an event for
@@ -1627,9 +1628,20 @@ macro_rules! handle_new_monitor_update {
 							})
 						} else { None }
 					} else { None };
-					let htlc_forwards = $self.handle_channel_resumption(&mut $msg_queue,
-						$chan, updates.raa, updates.commitment_update, updates.order, updates.accepted_htlcs,
-						updates.funding_broadcastable, updates.channel_ready, updates.announcement_sigs);
+
+					let htlc_forwards = {
+						let per_peer_state = $self.per_peer_state.read().unwrap();
+						let mut peer_state = per_peer_state.get(&counterparty_node_id)
+							.expect("XXX: This may be reachable today, I believe, but once we move the channel storage to per_peer_state it won't be.")
+							.lock().unwrap();
+
+						let update_actions = peer_state.monitor_update_blocked_actions
+							.remove(&$chan.channel_id()).unwrap_or(Vec::new());
+						$self.handle_channel_resumption(&mut $msg_queue, $chan,
+							update_actions, updates.raa, updates.commitment_update, updates.order,
+							updates.accepted_htlcs, updates.funding_broadcastable,
+							updates.channel_ready, updates.announcement_sigs)
+					};
 					if let Some(upd) = channel_update {
 						$msg_queue.push(upd);
 					}
@@ -4510,11 +4522,20 @@ impl<M: Deref, T: Deref, K: Deref, F: Deref, L: Deref> ChannelManager<M, T, K, F
 	/// Handles a channel reentering a functional state, either due to reconnect or a monitor
 	/// update completion.
 	fn handle_channel_resumption(&self, pending_msg_events: &mut Vec<MessageSendEvent>,
-		channel: &mut Channel<<K::Target as KeysInterface>::Signer>, raa: Option<msgs::RevokeAndACK>,
+		channel: &mut Channel<<K::Target as KeysInterface>::Signer>,
+		actions: Vec<MonitorUpdateCompletionAction>, raa: Option<msgs::RevokeAndACK>,
 		commitment_update: Option<msgs::CommitmentUpdate>, order: RAACommitmentOrder,
 		pending_forwards: Vec<(PendingHTLCInfo, u64)>, funding_broadcastable: Option<Transaction>,
 		channel_ready: Option<msgs::ChannelReady>, announcement_sigs: Option<msgs::AnnouncementSignatures>)
 	-> Option<(u64, OutPoint, u128, Vec<(PendingHTLCInfo, u64)>)> {
+		log_trace!(self.logger, "Handling channel resumption for channel {} with {} actions, {} RAA, {} commitment update, {} pending forwards, {}broadcasting funding, {} channel ready, {} announcement",
+			log_bytes!(channel.channel_id()), actions.len(),
+			if raa.is_some() { "an" } else { "no" },
+			if commitment_update.is_some() { "a" } else { "no" }, pending_forwards.len(),
+			if funding_broadcastable.is_some() { "" } else { "not " },
+			if channel_ready.is_some() { "sending" } else { "without" },
+			if announcement_sigs.is_some() { "sending" } else { "without" });
+
 		let mut htlc_forwards = None;
 
 		let counterparty_node_id = channel.get_counterparty_node_id();
@@ -4567,6 +4588,8 @@ impl<M: Deref, T: Deref, K: Deref, F: Deref, L: Deref> ChannelManager<M, T, K, F
 			self.tx_broadcaster.broadcast_transaction(&tx);
 		}
 
+		self.handle_monitor_update_completion_actions(actions);
+
 		htlc_forwards
 	}
 
@@ -4581,11 +4604,22 @@ impl<M: Deref, T: Deref, K: Deref, F: Deref, L: Deref> ChannelManager<M, T, K, F
 				hash_map::Entry::Occupied(chan) => chan,
 				hash_map::Entry::Vacant(_) => return,
 			};
+			log_trace!(self.logger, "ChannelMonitor updated to {}. Current highest is {}",
+				highest_applied_update_id, channel.get().get_latest_monitor_update_id());
 			if !channel.get().is_awaiting_monitor_update() || channel.get().get_latest_monitor_update_id() != highest_applied_update_id {
 				return;
 			}
 
 			let counterparty_node_id = channel.get().get_counterparty_node_id();
+			let per_peer_state = self.per_peer_state.read().unwrap();
+			let mut peer_state = match per_peer_state.get(&counterparty_node_id) {
+				Some(peer_state) => peer_state.lock().unwrap(),
+				None => {
+					debug_assert!(false, "XXX: This may be reachable today, I believe, but once we move the channel storage to per_peer_state it won't be.");
+					return;
+				}
+			};
+
 			let updates = channel.get_mut().monitor_updating_restored(&self.logger, self.get_our_node_id(), self.genesis_hash, self.best_block.read().unwrap().height());
 			let channel_update = if updates.channel_ready.is_some() && channel.get().is_usable() {
 				// We only send a channel_update in the case where we are just now sending a
@@ -4600,7 +4634,12 @@ impl<M: Deref, T: Deref, K: Deref, F: Deref, L: Deref> ChannelManager<M, T, K, F
 					})
 				} else { None }
 			} else { None };
-			htlc_forwards = self.handle_channel_resumption(&mut channel_state.pending_msg_events, channel.get_mut(), updates.raa, updates.commitment_update, updates.order, updates.accepted_htlcs, updates.funding_broadcastable, updates.channel_ready, updates.announcement_sigs);
+			let update_actions = peer_state.monitor_update_blocked_actions
+				.remove(&channel.get().channel_id()).unwrap_or(Vec::new());
+			htlc_forwards = self.handle_channel_resumption(&mut channel_state.pending_msg_events,
+				channel.get_mut(), update_actions,
+				updates.raa, updates.commitment_update, updates.order, updates.accepted_htlcs,
+				updates.funding_broadcastable, updates.channel_ready, updates.announcement_sigs);
 			if let Some(upd) = channel_update {
 				channel_state.pending_msg_events.push(upd);
 			}
@@ -5452,8 +5491,9 @@ impl<M: Deref, T: Deref, K: Deref, F: Deref, L: Deref> ChannelManager<M, T, K, F
 					}
 					let need_lnd_workaround = chan.get_mut().workaround_lnd_bug_4006.take();
 					htlc_forwards = self.handle_channel_resumption(
-						&mut channel_state.pending_msg_events, chan.get_mut(), responses.raa, responses.commitment_update, responses.order,
-						Vec::new(), None, responses.channel_ready, responses.announcement_sigs);
+						&mut channel_state.pending_msg_events, chan.get_mut(), Vec::new(),
+						responses.raa, responses.commitment_update, responses.order, Vec::new(),
+						None, responses.channel_ready, responses.announcement_sigs);
 					if let Some(upd) = channel_update {
 						channel_state.pending_msg_events.push(upd);
 					}