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

@@ -2587,7 +2587,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

@@ -3161,16 +3161,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(),
@@ -3251,16 +3251,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);
@@ -3269,16 +3269,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())
 		}
 	}
 
@@ -3488,17 +3483,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;
@@ -5815,8 +5802,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 {
@@ -5859,12 +5849,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,
@@ -5885,6 +5878,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

@@ -1947,7 +1947,8 @@ impl<M: Deref, T: Deref, K: Deref, F: Deref, L: Deref> ChannelManager<M, T, K, F
 					if *counterparty_node_id != chan_entry.get().get_counterparty_node_id(){
 						return Err(APIError::APIMisuseError { err: "The passed counterparty_node_id doesn't match the channel's counterparty node_id".to_owned() });
 					}
-					let (shutdown_msg, monitor_update, htlcs) = {
+					let funding_txo_opt = chan_entry.get().get_funding_txo();
+					let (shutdown_msg, mut monitor_update_opt, htlcs) = {
 						let per_peer_state = self.per_peer_state.read().unwrap();
 						match per_peer_state.get(&counterparty_node_id) {
 							Some(peer_state) => {
@@ -1960,22 +1961,21 @@ impl<M: Deref, T: Deref, K: Deref, F: Deref, L: Deref> ChannelManager<M, T, K, F
 					};
 					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.
 					channel_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, channel_state_lock, channel_state.pending_msg_events, 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) {
@@ -5527,48 +5527,35 @@ impl<M: Deref, T: Deref, K: Deref, F: Deref, L: Deref> ChannelManager<M, T, K, F
 		let mut has_monitor_update = false;
 		let mut failed_htlcs = Vec::new();
 		let mut handle_errors = Vec::new();
-		{
+		'outer_loop: loop {
 			let mut channel_state_lock = self.channel_state.lock().unwrap();
 			let channel_state = &mut *channel_state_lock;
-			let by_id = &mut channel_state.by_id;
-			let pending_msg_events = &mut channel_state.pending_msg_events;
 
-			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
+			for (channel_id, chan) in channel_state.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,
+						channel_state, channel_state.pending_msg_events, chan, MANUALLY_REMOVING,
+						channel_state.by_id.remove(&channel_id));
+					if res.is_err() {
+						handle_errors.push((counterparty_node_id, res));
 					}
+					continue 'outer_loop;
 				}
-			});
+			}
+			break 'outer_loop;
 		}
 
 		let has_update = has_monitor_update || !failed_htlcs.is_empty() || !handle_errors.is_empty();