Always process payment send 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 for
HTLC sends originating from an outbound payment.

Author: TheBlueMatt

lightning/src/ln/channel.rs

@@ -5635,15 +5635,6 @@ impl<Signer: Sign> Channel<Signer> {
 		Ok(Some(res))
 	}
 
-	/// Only fails in case of signer rejection.
-	fn send_commitment_no_status_check<L: Deref>(&mut self, logger: &L) -> Result<(msgs::CommitmentSigned, ChannelMonitorUpdate), ChannelError> where L::Target: Logger {
-		let monitor_update = self.build_commitment_no_status_check(logger);
-		match self.send_commitment_no_state_update(logger) {
-			Ok((commitment_signed, _)) => Ok((commitment_signed, monitor_update)),
-			Err(e) => Err(e),
-		}
-	}
-
 	fn build_commitment_no_status_check<L: Deref>(&mut self, logger: &L) -> ChannelMonitorUpdate where L::Target: Logger {
 		log_trace!(logger, "Updating HTLC state for a newly-sent commitment_signed...");
 		// We can upgrade the status of some HTLCs that are waiting on a commitment, even if we
@@ -5770,16 +5761,20 @@ impl<Signer: Sign> Channel<Signer> {
 		}, (counterparty_commitment_txid, commitment_stats.htlcs_included)))
 	}
 
-	/// Adds a pending outbound HTLC to this channel, and creates a signed commitment transaction
-	/// to send to the remote peer in one go.
+	/// Adds a pending outbound HTLC to this channel, and builds a new remote commitment
+	/// transaction and generates the corresponding [`ChannelMonitorUpdate`] in one go.
 	///
 	/// Shorthand for calling [`Self::send_htlc`] followed by a commitment update, see docs on
-	/// [`Self::send_htlc`] and [`Self::send_commitment_no_state_update`] for more info.
-	pub fn send_htlc_and_commit<L: Deref>(&mut self, amount_msat: u64, payment_hash: PaymentHash, cltv_expiry: u32, source: HTLCSource, onion_routing_packet: msgs::OnionPacket, logger: &L) -> Result<Option<(msgs::UpdateAddHTLC, msgs::CommitmentSigned, ChannelMonitorUpdate)>, ChannelError> where L::Target: Logger {
-		match self.send_htlc(amount_msat, payment_hash, cltv_expiry, source, onion_routing_packet, false, logger)? {
-			Some(update_add_htlc) => {
-				let (commitment_signed, monitor_update) = self.send_commitment_no_status_check(logger)?;
-				Ok(Some((update_add_htlc, commitment_signed, monitor_update)))
+	/// [`Self::send_htlc`] and [`Self::build_commitment_no_state_update`] for more info.
+	pub fn send_htlc_and_commit<L: Deref>(&mut self, amount_msat: u64, payment_hash: PaymentHash, cltv_expiry: u32, source: HTLCSource, onion_routing_packet: msgs::OnionPacket, logger: &L) -> Result<Option<&ChannelMonitorUpdate>, ChannelError> where L::Target: Logger {
+		let send_res = self.send_htlc(amount_msat, payment_hash, cltv_expiry, source, onion_routing_packet, false, logger);
+		if let Err(e) = &send_res { if let ChannelError::Ignore(_) = e {} else { debug_assert!(false, "Sending cannot trigger channel failure"); } }
+		match send_res? {
+			Some(_) => {
+				let monitor_update = self.build_commitment_no_status_check(logger);
+				self.monitor_updating_paused(false, true, false, Vec::new(), Vec::new(), Vec::new());
+				self.pending_monitor_updates.push(monitor_update);
+				Ok(Some(self.pending_monitor_updates.last().unwrap()))
 			},
 			None => Ok(None)
 		}

lightning/src/ln/channelmanager.rs

@@ -2550,6 +2550,7 @@ impl<M: Deref, T: Deref, K: Deref, F: Deref, L: Deref> ChannelManager<M, T, K, F
 					if !chan.get().is_live() {
 						return Err(APIError::ChannelUnavailable{err: "Peer for first hop currently disconnected/pending monitor update!".to_owned()});
 					}
+					(chan.get().get_funding_txo().unwrap(),
 					break_chan_entry!(self, chan.get_mut().send_htlc_and_commit(
 						htlc_msat, payment_hash.clone(), htlc_cltv, HTLCSource::OutboundRoute {
 							path: path.clone(),
@@ -2560,42 +2561,25 @@ impl<M: Deref, T: Deref, K: Deref, F: Deref, L: Deref> ChannelManager<M, T, K, F
 							payment_params: payment_params.clone(),
 						}, onion_packet, &self.logger),
 						chan)
+					)
 				} {
-					Some((update_add, commitment_signed, monitor_update)) => {
-						let update_err = self.chain_monitor.update_channel(chan.get().get_funding_txo().unwrap(), &monitor_update);
-						let chan_id = chan.get().channel_id();
-						match (update_err,
-							handle_monitor_update_res!(self, update_err, chan,
-								RAACommitmentOrder::CommitmentFirst, false, true))
-						{
-							(ChannelMonitorUpdateStatus::PermanentFailure, Err(e)) => break Err(e),
-							(ChannelMonitorUpdateStatus::Completed, Ok(())) => {},
-							(ChannelMonitorUpdateStatus::InProgress, Err(_)) => {
-								// Note that MonitorUpdateInProgress here indicates (per function
-								// docs) that we will resend the commitment update once monitor
-								// updating completes. Therefore, we must return an error
-								// indicating that it is unsafe to retry the payment wholesale,
-								// which we do in the send_payment check for
-								// MonitorUpdateInProgress, below.
-								return Err(APIError::MonitorUpdateInProgress);
-							},
-							_ => unreachable!(),
+					(funding_txo, Some(monitor_update)) => {
+						let update_id = monitor_update.update_id;
+						let update_res = self.chain_monitor.update_channel(funding_txo, monitor_update);
+						if let Err(e) = handle_new_monitor_update!(self, update_res, update_id, channel_lock, channel_state.pending_msg_events, chan) {
+							break Err(e);
+						}
+						if update_res == ChannelMonitorUpdateStatus::InProgress {
+							// Note that MonitorUpdateInProgress here indicates (per function
+							// docs) that we will resend the commitment update once monitor
+							// updating completes. Therefore, we must return an error
+							// indicating that it is unsafe to retry the payment wholesale,
+							// which we do in the send_payment check for
+							// MonitorUpdateInProgress, below.
+							return Err(APIError::MonitorUpdateInProgress);
 						}
-
-						log_debug!(self.logger, "Sending payment along path resulted in a commitment_signed for channel {}", log_bytes!(chan_id));
-						channel_state.pending_msg_events.push(events::MessageSendEvent::UpdateHTLCs {
-							node_id: path.first().unwrap().pubkey,
-							updates: msgs::CommitmentUpdate {
-								update_add_htlcs: vec![update_add],
-								update_fulfill_htlcs: Vec::new(),
-								update_fail_htlcs: Vec::new(),
-								update_fail_malformed_htlcs: Vec::new(),
-								update_fee: None,
-								commitment_signed,
-							},
-						});
 					},
-					None => { },
+					(_, None) => { },
 				}
 			} else {
 				// The channel was likely removed after we fetched the id from the