(XXX: Squash all later commits into this) Always process RAA/CS 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
handling inbound `revoke_and_ack` and `commitment_signed` messages.

Author: TheBlueMatt

lightning/src/ln/chanmon_update_fail_tests.rs

@@ -143,7 +143,7 @@ fn test_monitor_and_persister_update_fail() {
 		let mut node_0_per_peer_lock;
 		let mut node_0_peer_state_lock;
 		let mut channel = get_channel_ref!(nodes[0], nodes[1], node_0_per_peer_lock, node_0_peer_state_lock, chan.2);
-		if let Ok((_, _, update)) = channel.commitment_signed(&updates.commitment_signed, &node_cfgs[0].logger) {
+		if let Ok(update) = channel.commitment_signed(&updates.commitment_signed, &node_cfgs[0].logger) {
 			// Check that even though the persister is returning a InProgress,
 			// because the update is bogus, ultimately the error that's returned
 			// should be a PermanentFailure.
@@ -1739,7 +1739,6 @@ fn test_monitor_update_on_pending_forwards() {
 	chanmon_cfgs[1].persister.set_update_ret(ChannelMonitorUpdateStatus::InProgress);
 	expect_pending_htlcs_forwardable_and_htlc_handling_failed!(nodes[1], vec![HTLCDestination::NextHopChannel { node_id: Some(nodes[2].node.get_our_node_id()), channel_id: chan_2.2 }]);
 	check_added_monitors!(nodes[1], 1);
-	assert!(nodes[1].node.get_and_clear_pending_msg_events().is_empty());
 
 	chanmon_cfgs[1].persister.set_update_ret(ChannelMonitorUpdateStatus::Completed);
 	let (outpoint, latest_update, _) = nodes[1].chain_monitor.latest_monitor_update_id.lock().unwrap().get(&chan_1.2).unwrap().clone();
@@ -1753,17 +1752,17 @@ fn test_monitor_update_on_pending_forwards() {
 
 	let events = nodes[0].node.get_and_clear_pending_events();
 	assert_eq!(events.len(), 3);
-	if let Event::PaymentPathFailed { payment_hash, payment_failed_permanently, .. } = events[0] {
+	if let Event::PaymentPathFailed { payment_hash, payment_failed_permanently, .. } = events[1] {
 		assert_eq!(payment_hash, payment_hash_1);
 		assert!(payment_failed_permanently);
 	} else { panic!("Unexpected event!"); }
-	match events[1] {
+	match events[2] {
 		Event::PaymentFailed { payment_hash, .. } => {
 			assert_eq!(payment_hash, payment_hash_1);
 		},
 		_ => panic!("Unexpected event"),
 	}
-	match events[2] {
+	match events[0] {
 		Event::PendingHTLCsForwardable { .. } => { },
 		_ => panic!("Unexpected event"),
 	};

lightning/src/ln/channel.rs

@@ -411,17 +411,6 @@ pub enum UpdateFulfillCommitFetch {
 	DuplicateClaim {},
 }
 
-/// The return value of `revoke_and_ack` on success, primarily updates to other channels or HTLC
-/// state.
-pub(super) struct RAAUpdates {
-	pub commitment_update: Option<msgs::CommitmentUpdate>,
-	pub accepted_htlcs: Vec<(PendingHTLCInfo, u64)>,
-	pub failed_htlcs: Vec<(HTLCSource, PaymentHash, HTLCFailReason)>,
-	pub finalized_claimed_htlcs: Vec<HTLCSource>,
-	pub monitor_update: ChannelMonitorUpdate,
-	pub holding_cell_failed_htlcs: Vec<(HTLCSource, PaymentHash)>,
-}
-
 /// The return value of `monitor_updating_restored`
 pub(super) struct MonitorRestoreUpdates {
 	pub raa: Option<msgs::RevokeAndACK>,
@@ -3049,17 +3038,17 @@ impl<Signer: WriteableEcdsaChannelSigner> Channel<Signer> {
 		Ok(())
 	}
 
-	pub fn commitment_signed<L: Deref>(&mut self, msg: &msgs::CommitmentSigned, logger: &L) -> Result<(msgs::RevokeAndACK, Option<msgs::CommitmentSigned>, ChannelMonitorUpdate), (Option<ChannelMonitorUpdate>, ChannelError)>
+	pub fn commitment_signed<L: Deref>(&mut self, msg: &msgs::CommitmentSigned, logger: &L) -> Result<&ChannelMonitorUpdate, ChannelError>
 		where L::Target: Logger
 	{
 		if (self.channel_state & (ChannelState::ChannelReady as u32)) != (ChannelState::ChannelReady as u32) {
-			return Err((None, ChannelError::Close("Got commitment signed message when channel was not in an operational state".to_owned())));
+			return Err(ChannelError::Close("Got commitment signed message when channel was not in an operational state".to_owned()));
 		}
 		if self.channel_state & (ChannelState::PeerDisconnected as u32) == ChannelState::PeerDisconnected as u32 {
-			return Err((None, ChannelError::Close("Peer sent commitment_signed when we needed a channel_reestablish".to_owned())));
+			return Err(ChannelError::Close("Peer sent commitment_signed when we needed a channel_reestablish".to_owned()));
 		}
 		if self.channel_state & BOTH_SIDES_SHUTDOWN_MASK == BOTH_SIDES_SHUTDOWN_MASK && self.last_sent_closing_fee.is_some() {
-			return Err((None, ChannelError::Close("Peer sent commitment_signed after we'd started exchanging closing_signeds".to_owned())));
+			return Err(ChannelError::Close("Peer sent commitment_signed after we'd started exchanging closing_signeds".to_owned()));
 		}
 
 		let funding_script = self.get_funding_redeemscript();
@@ -3077,7 +3066,7 @@ impl<Signer: WriteableEcdsaChannelSigner> Channel<Signer> {
 				log_bytes!(self.counterparty_funding_pubkey().serialize()), encode::serialize_hex(&bitcoin_tx.transaction),
 				log_bytes!(sighash[..]), encode::serialize_hex(&funding_script), log_bytes!(self.channel_id()));
 			if let Err(_) = self.secp_ctx.verify_ecdsa(&sighash, &msg.signature, &self.counterparty_funding_pubkey()) {
-				return Err((None, ChannelError::Close("Invalid commitment tx signature from peer".to_owned())));
+				return Err(ChannelError::Close("Invalid commitment tx signature from peer".to_owned()));
 			}
 			bitcoin_tx.txid
 		};
@@ -3092,7 +3081,7 @@ impl<Signer: WriteableEcdsaChannelSigner> Channel<Signer> {
 			debug_assert!(!self.is_outbound());
 			let counterparty_reserve_we_require_msat = self.holder_selected_channel_reserve_satoshis * 1000;
 			if commitment_stats.remote_balance_msat < commitment_stats.total_fee_sat * 1000 + counterparty_reserve_we_require_msat {
-				return Err((None, ChannelError::Close("Funding remote cannot afford proposed new fee".to_owned())));
+				return Err(ChannelError::Close("Funding remote cannot afford proposed new fee".to_owned()));
 			}
 		}
 		#[cfg(any(test, fuzzing))]
@@ -3114,7 +3103,7 @@ impl<Signer: WriteableEcdsaChannelSigner> Channel<Signer> {
 		}
 
 		if msg.htlc_signatures.len() != commitment_stats.num_nondust_htlcs {
-			return Err((None, ChannelError::Close(format!("Got wrong number of HTLC signatures ({}) from remote. It must be {}", msg.htlc_signatures.len(), commitment_stats.num_nondust_htlcs))));
+			return Err(ChannelError::Close(format!("Got wrong number of HTLC signatures ({}) from remote. It must be {}", msg.htlc_signatures.len(), commitment_stats.num_nondust_htlcs)));
 		}
 
 		// TODO: Sadly, we pass HTLCs twice to ChannelMonitor: once via the HolderCommitmentTransaction and once via the update
@@ -3132,7 +3121,7 @@ impl<Signer: WriteableEcdsaChannelSigner> Channel<Signer> {
 					log_bytes!(msg.htlc_signatures[idx].serialize_compact()[..]), log_bytes!(keys.countersignatory_htlc_key.serialize()),
 					encode::serialize_hex(&htlc_tx), log_bytes!(htlc_sighash[..]), encode::serialize_hex(&htlc_redeemscript), log_bytes!(self.channel_id()));
 				if let Err(_) = self.secp_ctx.verify_ecdsa(&htlc_sighash, &msg.htlc_signatures[idx], &keys.countersignatory_htlc_key) {
-					return Err((None, ChannelError::Close("Invalid HTLC tx signature from peer".to_owned())));
+					return Err(ChannelError::Close("Invalid HTLC tx signature from peer".to_owned()));
 				}
 				htlcs_and_sigs.push((htlc, Some(msg.htlc_signatures[idx]), source));
 			} else {
@@ -3148,10 +3137,8 @@ impl<Signer: WriteableEcdsaChannelSigner> Channel<Signer> {
 			self.counterparty_funding_pubkey()
 		);
 
-		let next_per_commitment_point = self.holder_signer.get_per_commitment_point(self.cur_holder_commitment_transaction_number - 1, &self.secp_ctx);
 		self.holder_signer.validate_holder_commitment(&holder_commitment_tx, commitment_stats.preimages)
-			.map_err(|_| (None, ChannelError::Close("Failed to validate our commitment".to_owned())))?;
-		let per_commitment_secret = self.holder_signer.release_commitment_secret(self.cur_holder_commitment_transaction_number + 1);
+			.map_err(|_| ChannelError::Close("Failed to validate our commitment".to_owned()))?;
 
 		// Update state now that we've passed all the can-fail calls...
 		let mut need_commitment = false;
@@ -3196,7 +3183,7 @@ impl<Signer: WriteableEcdsaChannelSigner> Channel<Signer> {
 
 		self.cur_holder_commitment_transaction_number -= 1;
 		// Note that if we need_commitment & !AwaitingRemoteRevoke we'll call
-		// send_commitment_no_status_check() next which will reset this to RAAFirst.
+		// build_commitment_no_status_check() next which will reset this to RAAFirst.
 		self.resend_order = RAACommitmentOrder::CommitmentFirst;
 
 		if (self.channel_state & ChannelState::MonitorUpdateInProgress as u32) != 0 {
@@ -3208,37 +3195,35 @@ impl<Signer: WriteableEcdsaChannelSigner> Channel<Signer> {
 				// the corresponding HTLC status updates so that get_last_commitment_update
 				// includes the right HTLCs.
 				self.monitor_pending_commitment_signed = true;
-				let (_, mut additional_update) = self.send_commitment_no_status_check(logger).map_err(|e| (None, e))?;
-				// send_commitment_no_status_check may bump latest_monitor_id but we want them to be
+				let mut additional_update = self.build_commitment_no_status_check(logger);
+				// build_commitment_no_status_check may bump latest_monitor_id but we want them to be
 				// strictly increasing by one, so decrement it here.
 				self.latest_monitor_update_id = monitor_update.update_id;
 				monitor_update.updates.append(&mut additional_update.updates);
 			}
 			log_debug!(logger, "Received valid commitment_signed from peer in channel {}, updated HTLC state but awaiting a monitor update resolution to reply.",
 				log_bytes!(self.channel_id));
-			return Err((Some(monitor_update), ChannelError::Ignore("Previous monitor update failure prevented generation of RAA".to_owned())));
+			self.pending_monitor_updates.push(monitor_update);
+			return Ok(self.pending_monitor_updates.last().unwrap());
 		}
 
-		let commitment_signed = if need_commitment && (self.channel_state & (ChannelState::AwaitingRemoteRevoke as u32)) == 0 {
+		let need_commitment_signed = if need_commitment && (self.channel_state & (ChannelState::AwaitingRemoteRevoke as u32)) == 0 {
 			// If we're AwaitingRemoteRevoke we can't send a new commitment here, but that's ok -
 			// we'll send one right away when we get the revoke_and_ack when we
 			// free_holding_cell_htlcs().
-			let (msg, mut additional_update) = self.send_commitment_no_status_check(logger).map_err(|e| (None, e))?;
-			// send_commitment_no_status_check may bump latest_monitor_id but we want them to be
+			let mut additional_update = self.build_commitment_no_status_check(logger);
+			// build_commitment_no_status_check may bump latest_monitor_id but we want them to be
 			// strictly increasing by one, so decrement it here.
 			self.latest_monitor_update_id = monitor_update.update_id;
 			monitor_update.updates.append(&mut additional_update.updates);
-			Some(msg)
-		} else { None };
+			true
+		} else { false };
 
 		log_debug!(logger, "Received valid commitment_signed from peer in channel {}, updating HTLC state and responding with{} a revoke_and_ack.",
-			log_bytes!(self.channel_id()), if commitment_signed.is_some() { " our own commitment_signed and" } else { "" });
-
-		Ok((msgs::RevokeAndACK {
-			channel_id: self.channel_id,
-			per_commitment_secret,
-			next_per_commitment_point,
-		}, commitment_signed, monitor_update))
+			log_bytes!(self.channel_id()), if need_commitment_signed { " our own commitment_signed and" } else { "" });
+		self.pending_monitor_updates.push(monitor_update);
+		self.monitor_updating_paused(true, need_commitment_signed, false, Vec::new(), Vec::new(), Vec::new());
+		return Ok(self.pending_monitor_updates.last().unwrap());
 	}
 
 	/// Public version of the below, checking relevant preconditions first.
@@ -3370,7 +3355,7 @@ impl<Signer: WriteableEcdsaChannelSigner> Channel<Signer> {
 	/// waiting on this revoke_and_ack. The generation of this new commitment_signed may also fail,
 	/// generating an appropriate error *after* the channel state has been updated based on the
 	/// revoke_and_ack message.
-	pub fn revoke_and_ack<L: Deref>(&mut self, msg: &msgs::RevokeAndACK, logger: &L) -> Result<RAAUpdates, ChannelError>
+	pub fn revoke_and_ack<L: Deref>(&mut self, msg: &msgs::RevokeAndACK, logger: &L) -> Result<(Vec<(HTLCSource, PaymentHash)>, &ChannelMonitorUpdate), ChannelError>
 		where L::Target: Logger,
 	{
 		if (self.channel_state & (ChannelState::ChannelReady as u32)) != (ChannelState::ChannelReady as u32) {
@@ -3557,8 +3542,8 @@ impl<Signer: WriteableEcdsaChannelSigner> Channel<Signer> {
 				// When the monitor updating is restored we'll call get_last_commitment_update(),
 				// which does not update state, but we're definitely now awaiting a remote revoke
 				// before we can step forward any more, so set it here.
-				let (_, mut additional_update) = self.send_commitment_no_status_check(logger)?;
-				// send_commitment_no_status_check may bump latest_monitor_id but we want them to be
+				let mut additional_update = self.build_commitment_no_status_check(logger);
+				// build_commitment_no_status_check may bump latest_monitor_id but we want them to be
 				// strictly increasing by one, so decrement it here.
 				self.latest_monitor_update_id = monitor_update.update_id;
 				monitor_update.updates.append(&mut additional_update.updates);
@@ -3567,12 +3552,8 @@ impl<Signer: WriteableEcdsaChannelSigner> Channel<Signer> {
 			self.monitor_pending_failures.append(&mut revoked_htlcs);
 			self.monitor_pending_finalized_fulfills.append(&mut finalized_claimed_htlcs);
 			log_debug!(logger, "Received a valid revoke_and_ack for channel {} but awaiting a monitor update resolution to reply.", log_bytes!(self.channel_id()));
-			return Ok(RAAUpdates {
-				commitment_update: None, finalized_claimed_htlcs: Vec::new(),
-				accepted_htlcs: Vec::new(), failed_htlcs: Vec::new(),
-				monitor_update,
-				holding_cell_failed_htlcs: Vec::new()
-			});
+			self.pending_monitor_updates.push(monitor_update);
+			return Ok((Vec::new(), self.pending_monitor_updates.last().unwrap()));
 		}
 
 		match self.free_holding_cell_htlcs(logger)? {
@@ -3591,47 +3572,29 @@ impl<Signer: WriteableEcdsaChannelSigner> Channel<Signer> {
 				self.latest_monitor_update_id = monitor_update.update_id;
 				monitor_update.updates.append(&mut additional_update.updates);
 
-				Ok(RAAUpdates {
-					commitment_update: Some(commitment_update),
-					finalized_claimed_htlcs,
-					accepted_htlcs: to_forward_infos,
-					failed_htlcs: revoked_htlcs,
-					monitor_update,
-					holding_cell_failed_htlcs: htlcs_to_fail
-				})
+				self.monitor_updating_paused(false, true, false, to_forward_infos, revoked_htlcs, finalized_claimed_htlcs);
+				self.pending_monitor_updates.push(monitor_update);
+				Ok((htlcs_to_fail, self.pending_monitor_updates.last().unwrap()))
 			},
 			(None, htlcs_to_fail) => {
 				if require_commitment {
-					let (commitment_signed, mut additional_update) = self.send_commitment_no_status_check(logger)?;
+					let mut additional_update = self.build_commitment_no_status_check(logger);
 
-					// send_commitment_no_status_check may bump latest_monitor_id but we want them to be
+					// build_commitment_no_status_check may bump latest_monitor_id but we want them to be
 					// strictly increasing by one, so decrement it here.
 					self.latest_monitor_update_id = monitor_update.update_id;
 					monitor_update.updates.append(&mut additional_update.updates);
 
 					log_debug!(logger, "Received a valid revoke_and_ack for channel {}. Responding with a commitment update with {} HTLCs failed.",
 						log_bytes!(self.channel_id()), update_fail_htlcs.len() + update_fail_malformed_htlcs.len());
-					Ok(RAAUpdates {
-						commitment_update: Some(msgs::CommitmentUpdate {
-							update_add_htlcs: Vec::new(),
-							update_fulfill_htlcs: Vec::new(),
-							update_fail_htlcs,
-							update_fail_malformed_htlcs,
-							update_fee: None,
-							commitment_signed
-						}),
-						finalized_claimed_htlcs,
-						accepted_htlcs: to_forward_infos, failed_htlcs: revoked_htlcs,
-						monitor_update, holding_cell_failed_htlcs: htlcs_to_fail
-					})
+					self.monitor_updating_paused(false, true, false, to_forward_infos, revoked_htlcs, finalized_claimed_htlcs);
+					self.pending_monitor_updates.push(monitor_update);
+					Ok((htlcs_to_fail, self.pending_monitor_updates.last().unwrap()))
 				} else {
 					log_debug!(logger, "Received a valid revoke_and_ack for channel {} with no reply necessary.", log_bytes!(self.channel_id()));
-					Ok(RAAUpdates {
-						commitment_update: None,
-						finalized_claimed_htlcs,
-						accepted_htlcs: to_forward_infos, failed_htlcs: revoked_htlcs,
-						monitor_update, holding_cell_failed_htlcs: htlcs_to_fail
-					})
+					self.monitor_updating_paused(false, false, false, to_forward_infos, revoked_htlcs, finalized_claimed_htlcs);
+					self.pending_monitor_updates.push(monitor_update);
+					Ok((htlcs_to_fail, self.pending_monitor_updates.last().unwrap()))
 				}
 			}
 		}
@@ -3818,6 +3781,7 @@ impl<Signer: WriteableEcdsaChannelSigner> Channel<Signer> {
 	{
 		assert_eq!(self.channel_state & ChannelState::MonitorUpdateInProgress as u32, ChannelState::MonitorUpdateInProgress as u32);
 		self.channel_state &= !(ChannelState::MonitorUpdateInProgress as u32);
+		self.pending_monitor_updates.clear();
 
 		// If we're past (or at) the FundingSent stage on an outbound channel, try to
 		// (re-)broadcast the funding transaction as we may have declined to broadcast it when we