Always process holding cell ChannelMonitorUpdates asynchronously

We currently have two codepaths on most channel update functions -
most methods return a set of messages to send a peer iff the
`ChannelMonitorUpdate` succeeds, but if it does not we push the
messages back into the `Channel` and then pull them back out when
the `ChannelMonitorUpdate` completes and send them then. This adds
a substantial amount of complexity in very critical codepaths.

Instead, here we swap all our channel update codepaths to
immediately set the channel-update-required flag and only return a
`ChannelMonitorUpdate` to the `ChannelManager`. Internally in the
`Channel` we store a queue of `ChannelMonitorUpdate`s, which will
become critical in future work to surface pending
`ChannelMonitorUpdate`s to users at startup so they can complete.

This leaves some redundant work in `Channel` to be cleaned up
later. Specifically, we still generate the messages which we will
now ignore and regenerate later.

This commit updates the `ChannelMonitorUpdate` pipeline when
freeing the holding cell, including when initiating shutdown.

Author: TheBlueMatt

lightning/src/ln/chanmon_update_fail_tests.rs

@@ -2600,7 +2600,15 @@ fn test_permanent_error_during_sending_shutdown() {
 	chanmon_cfgs[0].persister.set_update_ret(ChannelMonitorUpdateStatus::PermanentFailure);
 
 	assert!(nodes[0].node.close_channel(&channel_id, &nodes[1].node.get_our_node_id()).is_ok());
-	check_closed_broadcast!(nodes[0], true);
+
+	// We always send the `shutdown` response when initiating a shutdown, even if we immediately
+	// close the channel thereafter.
+	let msg_events = nodes[0].node.get_and_clear_pending_msg_events();
+	assert_eq!(msg_events.len(), 3);
+	if let MessageSendEvent::SendShutdown { .. } = msg_events[0] {} else { panic!(); }
+	if let MessageSendEvent::BroadcastChannelUpdate { .. } = msg_events[1] {} else { panic!(); }
+	if let MessageSendEvent::HandleError { .. } =  msg_events[2] {} else { panic!(); }
+
 	check_added_monitors!(nodes[0], 2);
 	check_closed_event!(nodes[0], 1, ClosureReason::ProcessingError { err: "ChannelMonitor storage failure".to_string() });
 }

lightning/src/ln/channel.rs

@@ -3186,16 +3186,16 @@ impl<Signer: Sign> Channel<Signer> {
 	/// Public version of the below, checking relevant preconditions first.
 	/// If we're not in a state where freeing the holding cell makes sense, this is a no-op and
 	/// returns `(None, Vec::new())`.
-	pub fn maybe_free_holding_cell_htlcs<L: Deref>(&mut self, logger: &L) -> Result<(Option<(msgs::CommitmentUpdate, ChannelMonitorUpdate)>, Vec<(HTLCSource, PaymentHash)>), ChannelError> where L::Target: Logger {
+	pub fn maybe_free_holding_cell_htlcs<L: Deref>(&mut self, logger: &L) -> (Option<&ChannelMonitorUpdate>, Vec<(HTLCSource, PaymentHash)>) where L::Target: Logger {
 		if self.channel_state >= ChannelState::ChannelReady as u32 &&
 		   (self.channel_state & (ChannelState::AwaitingRemoteRevoke as u32 | ChannelState::PeerDisconnected as u32 | ChannelState::MonitorUpdateInProgress as u32)) == 0 {
 			self.free_holding_cell_htlcs(logger)
-		} else { Ok((None, Vec::new())) }
+		} else { (None, Vec::new()) }
 	}
 
 	/// Frees any pending commitment updates in the holding cell, generating the relevant messages
 	/// for our counterparty.
-	fn free_holding_cell_htlcs<L: Deref>(&mut self, logger: &L) -> Result<(Option<(msgs::CommitmentUpdate, ChannelMonitorUpdate)>, Vec<(HTLCSource, PaymentHash)>), ChannelError> where L::Target: Logger {
+	fn free_holding_cell_htlcs<L: Deref>(&mut self, logger: &L) -> (Option<&ChannelMonitorUpdate>, Vec<(HTLCSource, PaymentHash)>) where L::Target: Logger {
 		assert_eq!(self.channel_state & ChannelState::MonitorUpdateInProgress as u32, 0);
 		if self.holding_cell_htlc_updates.len() != 0 || self.holding_cell_update_fee.is_some() {
 			log_trace!(logger, "Freeing holding cell with {} HTLC updates{} in channel {}", self.holding_cell_htlc_updates.len(),
@@ -3276,16 +3276,16 @@ impl<Signer: Sign> Channel<Signer> {
 				}
 			}
 			if update_add_htlcs.is_empty() && update_fulfill_htlcs.is_empty() && update_fail_htlcs.is_empty() && self.holding_cell_update_fee.is_none() {
-				return Ok((None, htlcs_to_fail));
+				return (None, htlcs_to_fail);
 			}
 			let update_fee = if let Some(feerate) = self.holding_cell_update_fee.take() {
 				self.send_update_fee(feerate, false, logger)
 			} else {
 				None
 			};
 
-			let (commitment_signed, mut additional_update) = self.send_commitment_no_status_check(logger)?;
-			// send_commitment_no_status_check and get_update_fulfill_htlc may bump latest_monitor_id
+			let mut additional_update = self.build_commitment_no_status_check(logger);
+			// build_commitment_no_status_check and get_update_fulfill_htlc may bump latest_monitor_id
 			// but we want them to be strictly increasing by one, so reset it here.
 			self.latest_monitor_update_id = monitor_update.update_id;
 			monitor_update.updates.append(&mut additional_update.updates);
@@ -3294,16 +3294,11 @@ impl<Signer: Sign> Channel<Signer> {
 				log_bytes!(self.channel_id()), if update_fee.is_some() { "a fee update, " } else { "" },
 				update_add_htlcs.len(), update_fulfill_htlcs.len(), update_fail_htlcs.len());
 
-			Ok((Some((msgs::CommitmentUpdate {
-				update_add_htlcs,
-				update_fulfill_htlcs,
-				update_fail_htlcs,
-				update_fail_malformed_htlcs: Vec::new(),
-				update_fee,
-				commitment_signed,
-			}, monitor_update)), htlcs_to_fail))
+			self.monitor_updating_paused(false, true, false, Vec::new(), Vec::new(), Vec::new());
+			self.pending_monitor_updates.push(monitor_update);
+			(Some(self.pending_monitor_updates.last().unwrap()), htlcs_to_fail)
 		} else {
-			Ok((None, Vec::new()))
+			(None, Vec::new())
 		}
 	}
 
@@ -3513,17 +3508,9 @@ impl<Signer: Sign> Channel<Signer> {
 			return Ok((Vec::new(), self.pending_monitor_updates.last().unwrap()));
 		}
 
-		match self.free_holding_cell_htlcs(logger)? {
-			(Some((mut commitment_update, mut additional_update)), htlcs_to_fail) => {
-				commitment_update.update_fail_htlcs.reserve(update_fail_htlcs.len());
-				for fail_msg in update_fail_htlcs.drain(..) {
-					commitment_update.update_fail_htlcs.push(fail_msg);
-				}
-				commitment_update.update_fail_malformed_htlcs.reserve(update_fail_malformed_htlcs.len());
-				for fail_msg in update_fail_malformed_htlcs.drain(..) {
-					commitment_update.update_fail_malformed_htlcs.push(fail_msg);
-				}
-
+		match self.free_holding_cell_htlcs(logger) {
+			(Some(_), htlcs_to_fail) => {
+				let mut additional_update = self.pending_monitor_updates.pop().unwrap();
 				// free_holding_cell_htlcs may bump latest_monitor_id multiple times but we want them to be
 				// strictly increasing by one, so decrement it here.
 				self.latest_monitor_update_id = monitor_update.update_id;
@@ -5843,8 +5830,11 @@ impl<Signer: Sign> Channel<Signer> {
 
 	/// Begins the shutdown process, getting a message for the remote peer and returning all
 	/// holding cell HTLCs for payment failure.
+	///
+	/// May jump to the channel being fully shutdown (see [`Self::is_shutdown`]) in which case no
+	/// [`ChannelMonitorUpdate`] will be returned).
 	pub fn get_shutdown<K: Deref>(&mut self, keys_provider: &K, their_features: &InitFeatures, target_feerate_sats_per_kw: Option<u32>)
-	-> Result<(msgs::Shutdown, Option<ChannelMonitorUpdate>, Vec<(HTLCSource, PaymentHash)>), APIError>
+	-> Result<(msgs::Shutdown, Option<&ChannelMonitorUpdate>, Vec<(HTLCSource, PaymentHash)>), APIError>
 	where K::Target: KeysInterface {
 		for htlc in self.pending_outbound_htlcs.iter() {
 			if let OutboundHTLCState::LocalAnnounced(_) = htlc.state {
@@ -5887,12 +5877,15 @@ impl<Signer: Sign> Channel<Signer> {
 
 		let monitor_update = if update_shutdown_script {
 			self.latest_monitor_update_id += 1;
-			Some(ChannelMonitorUpdate {
+			let monitor_update = ChannelMonitorUpdate {
 				update_id: self.latest_monitor_update_id,
 				updates: vec![ChannelMonitorUpdateStep::ShutdownScript {
 					scriptpubkey: self.get_closing_scriptpubkey(),
 				}],
-			})
+			};
+			self.monitor_updating_paused(false, false, false, Vec::new(), Vec::new(), Vec::new());
+			self.pending_monitor_updates.push(monitor_update);
+			Some(self.pending_monitor_updates.last().unwrap())
 		} else { None };
 		let shutdown = msgs::Shutdown {
 			channel_id: self.channel_id,
@@ -5913,6 +5906,9 @@ impl<Signer: Sign> Channel<Signer> {
 			}
 		});
 
+		debug_assert!(!self.is_shutdown() || monitor_update.is_none(),
+			"we can't both complete shutdown and return a monitor update");
+
 		Ok((shutdown, monitor_update, dropped_outbound_htlcs))
 	}
 

lightning/src/ln/channelmanager.rs

@@ -1774,28 +1774,30 @@ where
 			}
 
 			let mut peer_state_lock = peer_state_mutex_opt.unwrap().lock().unwrap();
-			let peer_state = &mut *peer_state_lock;
+			let peer_state: &mut PeerState<_> = &mut *peer_state_lock;
 			match peer_state.channel_by_id.entry(channel_id.clone()) {
 				hash_map::Entry::Occupied(mut chan_entry) => {
-					let (shutdown_msg, monitor_update, htlcs) = chan_entry.get_mut().get_shutdown(&self.keys_manager, &peer_state.latest_features, target_feerate_sats_per_1000_weight)?;
+					let funding_txo_opt = chan_entry.get().get_funding_txo();
+					let their_features = &peer_state.latest_features;
+					let (shutdown_msg, mut monitor_update_opt, htlcs) = chan_entry.get_mut()
+						.get_shutdown(&self.keys_manager, their_features, target_feerate_sats_per_1000_weight)?;
 					failed_htlcs = htlcs;
 
-					// Update the monitor with the shutdown script if necessary.
-					if let Some(monitor_update) = monitor_update {
-						let update_res = self.chain_monitor.update_channel(chan_entry.get().get_funding_txo().unwrap(), &monitor_update);
-						let (result, is_permanent) =
-							handle_monitor_update_res!(self, update_res, chan_entry.get_mut(), RAACommitmentOrder::CommitmentFirst, chan_entry.key(), NO_UPDATE);
-						if is_permanent {
-							remove_channel!(self, chan_entry);
-							break result;
-						}
-					}
-
+					// We can send the `shutdown` message before updating the `ChannelMonitor`
+					// here as we don't need the monitor update to complete until we send a
+					// `shutdown_signed`, which we'll delay if we're pending a monitor update.
 					peer_state.pending_msg_events.push(events::MessageSendEvent::SendShutdown {
 						node_id: *counterparty_node_id,
-						msg: shutdown_msg
+						msg: shutdown_msg,
 					});
 
+					// Update the monitor with the shutdown script if necessary.
+					if let Some(monitor_update) = monitor_update_opt.take() {
+						let update_id = monitor_update.update_id;
+						let update_res = self.chain_monitor.update_channel(funding_txo_opt.unwrap(), monitor_update);
+						break handle_new_monitor_update!(self, update_res, update_id, peer_state_lock, peer_state, chan_entry);
+					}
+
 					if chan_entry.get().is_shutdown() {
 						let channel = remove_channel!(self, chan_entry);
 						if let Ok(channel_update) = self.get_channel_update_for_broadcast(&channel) {
@@ -5060,49 +5062,37 @@ where
 		let mut has_monitor_update = false;
 		let mut failed_htlcs = Vec::new();
 		let mut handle_errors = Vec::new();
-		{
-			let per_peer_state = self.per_peer_state.read().unwrap();
+		let per_peer_state = self.per_peer_state.read().unwrap();
 
-			for (_cp_id, peer_state_mutex) in per_peer_state.iter() {
+		for (_cp_id, peer_state_mutex) in per_peer_state.iter() {
+			'chan_loop: loop {
 				let mut peer_state_lock = peer_state_mutex.lock().unwrap();
-				let peer_state = &mut *peer_state_lock;
-				let pending_msg_events = &mut peer_state.pending_msg_events;
-				peer_state.channel_by_id.retain(|channel_id, chan| {
-					match chan.maybe_free_holding_cell_htlcs(&self.logger) {
-						Ok((commitment_opt, holding_cell_failed_htlcs)) => {
-							if !holding_cell_failed_htlcs.is_empty() {
-								failed_htlcs.push((
-									holding_cell_failed_htlcs,
-									*channel_id,
-									chan.get_counterparty_node_id()
-								));
-							}
-							if let Some((commitment_update, monitor_update)) = commitment_opt {
-								match self.chain_monitor.update_channel(chan.get_funding_txo().unwrap(), &monitor_update) {
-									ChannelMonitorUpdateStatus::Completed => {
-										pending_msg_events.push(events::MessageSendEvent::UpdateHTLCs {
-											node_id: chan.get_counterparty_node_id(),
-											updates: commitment_update,
-										});
-									},
-									e => {
-										has_monitor_update = true;
-										let (res, close_channel) = handle_monitor_update_res!(self, e, chan, RAACommitmentOrder::CommitmentFirst, channel_id, COMMITMENT_UPDATE_ONLY);
-										handle_errors.push((chan.get_counterparty_node_id(), res));
-										if close_channel { return false; }
-									},
-								}
-							}
-							true
-						},
-						Err(e) => {
-							let (close_channel, res) = convert_chan_err!(self, e, chan, channel_id);
-							handle_errors.push((chan.get_counterparty_node_id(), Err(res)));
-							// ChannelClosed event is generated by handle_error for us
-							!close_channel
+				let peer_state: &mut PeerState<_> = &mut *peer_state_lock;
+				for (channel_id, chan) in peer_state.channel_by_id.iter_mut() {
+					let counterparty_node_id = chan.get_counterparty_node_id();
+					let funding_txo = chan.get_funding_txo();
+					let (monitor_opt, holding_cell_failed_htlcs) =
+						chan.maybe_free_holding_cell_htlcs(&self.logger);
+					if !holding_cell_failed_htlcs.is_empty() {
+						failed_htlcs.push((holding_cell_failed_htlcs, *channel_id, counterparty_node_id));
+					}
+					if let Some(monitor_update) = monitor_opt {
+						has_monitor_update = true;
+
+						let update_res = self.chain_monitor.update_channel(
+							funding_txo.expect("channel is live"), monitor_update);
+						let update_id = monitor_update.update_id;
+						let channel_id: [u8; 32] = *channel_id;
+						let res = handle_new_monitor_update!(self, update_res, update_id,
+							peer_state_lock, peer_state, chan, MANUALLY_REMOVING,
+							peer_state.channel_by_id.remove(&channel_id));
+						if res.is_err() {
+							handle_errors.push((counterparty_node_id, res));
 						}
+						continue 'chan_loop;
 					}
-				});
+				}
+				break 'chan_loop;
 			}
 		}