Always process claim 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
`claim_funds_from_hop`, ie the `update_fulfill_htlc`-generation
pipeline.

Author: TheBlueMatt

lightning/src/chain/chainmonitor.rs

@@ -796,7 +796,7 @@ mod tests {
 	use crate::ln::functional_test_utils::*;
 	use crate::ln::msgs::ChannelMessageHandler;
 	use crate::util::errors::APIError;
-	use crate::util::events::{ClosureReason, MessageSendEvent, MessageSendEventsProvider};
+	use crate::util::events::{Event, ClosureReason, MessageSendEvent, MessageSendEventsProvider};
 
 	#[test]
 	fn test_async_ooo_offchain_updates() {
@@ -818,10 +818,8 @@ mod tests {
 
 		nodes[1].node.claim_funds(payment_preimage_1);
 		check_added_monitors!(nodes[1], 1);
-		expect_payment_claimed!(nodes[1], payment_hash_1, 1_000_000);
 		nodes[1].node.claim_funds(payment_preimage_2);
 		check_added_monitors!(nodes[1], 1);
-		expect_payment_claimed!(nodes[1], payment_hash_2, 1_000_000);
 
 		chanmon_cfgs[1].persister.set_update_ret(ChannelMonitorUpdateStatus::Completed);
 
@@ -851,8 +849,24 @@ mod tests {
 			.find(|(txo, _)| txo == funding_txo).unwrap().1.contains(&next_update));
 		assert!(nodes[1].chain_monitor.release_pending_monitor_events().is_empty());
 		assert!(nodes[1].node.get_and_clear_pending_msg_events().is_empty());
+		assert!(nodes[1].node.get_and_clear_pending_events().is_empty());
 		nodes[1].chain_monitor.chain_monitor.channel_monitor_updated(*funding_txo, update_iter.next().unwrap().clone()).unwrap();
 
+		let claim_events = nodes[1].node.get_and_clear_pending_events();
+		assert_eq!(claim_events.len(), 2);
+		match claim_events[0] {
+			Event::PaymentClaimed { ref payment_hash, amount_msat: 1_000_000, .. } => {
+				assert_eq!(payment_hash_1, *payment_hash);
+			},
+			_ => panic!("Unexpected event"),
+		}
+		match claim_events[1] {
+			Event::PaymentClaimed { ref payment_hash, amount_msat: 1_000_000, .. } => {
+				assert_eq!(payment_hash_2, *payment_hash);
+			},
+			_ => panic!("Unexpected event"),
+		}
+
 		// Now manually walk the commitment signed dance - because we claimed two payments
 		// back-to-back it doesn't fit into the neat walk commitment_signed_dance does.
 

lightning/src/ln/chanmon_update_fail_tests.rs

@@ -1590,7 +1590,6 @@ fn test_monitor_update_fail_claim() {
 
 	chanmon_cfgs[1].persister.set_update_ret(ChannelMonitorUpdateStatus::InProgress);
 	nodes[1].node.claim_funds(payment_preimage_1);
-	expect_payment_claimed!(nodes[1], payment_hash_1, 1_000_000);
 	assert!(nodes[1].node.get_and_clear_pending_msg_events().is_empty());
 	check_added_monitors!(nodes[1], 1);
 
@@ -1616,6 +1615,7 @@ fn test_monitor_update_fail_claim() {
 	let events = nodes[1].node.get_and_clear_pending_msg_events();
 	assert_eq!(events.len(), 0);
 	commitment_signed_dance!(nodes[1], nodes[2], payment_event.commitment_msg, false, true);
+	expect_pending_htlcs_forwardable_ignore!(nodes[1]);
 
 	let (_, payment_hash_3, payment_secret_3) = get_payment_preimage_hash!(nodes[0]);
 	nodes[2].node.send_payment(&route, payment_hash_3, &Some(payment_secret_3), PaymentId(payment_hash_3.0)).unwrap();
@@ -1633,6 +1633,7 @@ fn test_monitor_update_fail_claim() {
 	let channel_id = chan_1.2;
 	let (outpoint, latest_update, _) = nodes[1].chain_monitor.latest_monitor_update_id.lock().unwrap().get(&channel_id).unwrap().clone();
 	nodes[1].chain_monitor.chain_monitor.force_channel_monitor_updated(outpoint, latest_update);
+	expect_payment_claimed!(nodes[1], payment_hash_1, 1_000_000);
 	check_added_monitors!(nodes[1], 0);
 
 	let bs_fulfill_update = get_htlc_update_msgs!(nodes[1], nodes[0].node.get_our_node_id());
@@ -1641,7 +1642,7 @@ fn test_monitor_update_fail_claim() {
 	expect_payment_sent!(nodes[0], payment_preimage_1);
 
 	// Get the payment forwards, note that they were batched into one commitment update.
-	expect_pending_htlcs_forwardable!(nodes[1]);
+	nodes[1].node.process_pending_htlc_forwards();
 	check_added_monitors!(nodes[1], 1);
 	let bs_forward_update = get_htlc_update_msgs!(nodes[1], nodes[0].node.get_our_node_id());
 	nodes[0].node.handle_update_add_htlc(&nodes[1].node.get_our_node_id(), &bs_forward_update.update_add_htlcs[0]);
@@ -1784,14 +1785,14 @@ fn monitor_update_claim_fail_no_response() {
 
 	chanmon_cfgs[1].persister.set_update_ret(ChannelMonitorUpdateStatus::InProgress);
 	nodes[1].node.claim_funds(payment_preimage_1);
-	expect_payment_claimed!(nodes[1], payment_hash_1, 1_000_000);
 	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(&channel_id).unwrap().clone();
 	nodes[1].chain_monitor.chain_monitor.force_channel_monitor_updated(outpoint, latest_update);
+	expect_payment_claimed!(nodes[1], payment_hash_1, 1_000_000);
 	check_added_monitors!(nodes[1], 0);
 	assert!(nodes[1].node.get_and_clear_pending_msg_events().is_empty());
 
@@ -2270,7 +2271,6 @@ fn do_channel_holding_cell_serialize(disconnect: bool, reload_a: bool) {
 	chanmon_cfgs[0].persister.set_update_ret(ChannelMonitorUpdateStatus::InProgress);
 	nodes[0].node.claim_funds(payment_preimage_0);
 	check_added_monitors!(nodes[0], 1);
-	expect_payment_claimed!(nodes[0], payment_hash_0, 100_000);
 
 	nodes[1].node.handle_update_add_htlc(&nodes[0].node.get_our_node_id(), &send.msgs[0]);
 	nodes[1].node.handle_commitment_signed(&nodes[0].node.get_our_node_id(), &send.commitment_msg);
@@ -2333,6 +2333,7 @@ fn do_channel_holding_cell_serialize(disconnect: bool, reload_a: bool) {
 	chanmon_cfgs[0].persister.set_update_ret(ChannelMonitorUpdateStatus::Completed);
 	let (funding_txo, mon_id, _) = nodes[0].chain_monitor.latest_monitor_update_id.lock().unwrap().get(&chan_id).unwrap().clone();
 	nodes[0].chain_monitor.chain_monitor.force_channel_monitor_updated(funding_txo, mon_id);
+	expect_payment_claimed!(nodes[0], payment_hash_0, 100_000);
 
 	// New outbound messages should be generated immediately upon a call to
 	// get_and_clear_pending_msg_events (but not before).
@@ -2631,15 +2632,13 @@ fn double_temp_error() {
 	// `claim_funds` results in a ChannelMonitorUpdate.
 	nodes[1].node.claim_funds(payment_preimage_1);
 	check_added_monitors!(nodes[1], 1);
-	expect_payment_claimed!(nodes[1], payment_hash_1, 1_000_000);
 	let (funding_tx, latest_update_1, _) = nodes[1].chain_monitor.latest_monitor_update_id.lock().unwrap().get(&channel_id).unwrap().clone();
 
 	chanmon_cfgs[1].persister.set_update_ret(ChannelMonitorUpdateStatus::InProgress);
 	// Previously, this would've panicked due to a double-call to `Channel::monitor_update_failed`,
 	// which had some asserts that prevented it from being called twice.
 	nodes[1].node.claim_funds(payment_preimage_2);
 	check_added_monitors!(nodes[1], 1);
-	expect_payment_claimed!(nodes[1], payment_hash_2, 1_000_000);
 	chanmon_cfgs[1].persister.set_update_ret(ChannelMonitorUpdateStatus::Completed);
 
 	let (_, latest_update_2, _) = nodes[1].chain_monitor.latest_monitor_update_id.lock().unwrap().get(&channel_id).unwrap().clone();
@@ -2648,11 +2647,24 @@ fn double_temp_error() {
 	check_added_monitors!(nodes[1], 0);
 	nodes[1].chain_monitor.chain_monitor.force_channel_monitor_updated(funding_tx, latest_update_2);
 
-	// Complete the first HTLC.
-	let events = nodes[1].node.get_and_clear_pending_msg_events();
-	assert_eq!(events.len(), 1);
+	// Complete the first HTLC. Note that as a side-effect we handle the monitor update completions
+	// and get both PaymentClaimed events at once.
+	let msg_events = nodes[1].node.get_and_clear_pending_msg_events();
+
+	let events = nodes[1].node.get_and_clear_pending_events();
+	assert_eq!(events.len(), 2);
+	match events[0] {
+		Event::PaymentClaimed { amount_msat: 1_000_000, payment_hash, .. } => assert_eq!(payment_hash, payment_hash_1),
+		_ => panic!("Unexpected Event: {:?}", events[0]),
+	}
+	match events[1] {
+		Event::PaymentClaimed { amount_msat: 1_000_000, payment_hash, .. } => assert_eq!(payment_hash, payment_hash_2),
+		_ => panic!("Unexpected Event: {:?}", events[1]),
+	}
+
+	assert_eq!(msg_events.len(), 1);
 	let (update_fulfill_1, commitment_signed_b1, node_id) = {
-		match &events[0] {
+		match &msg_events[0] {
 			&MessageSendEvent::UpdateHTLCs { ref node_id, updates: msgs::CommitmentUpdate { ref update_add_htlcs, ref update_fulfill_htlcs, ref update_fail_htlcs, ref update_fail_malformed_htlcs, ref update_fee, ref commitment_signed } } => {
 				assert!(update_add_htlcs.is_empty());
 				assert_eq!(update_fulfill_htlcs.len(), 1);

lightning/src/ln/channel.rs

@@ -391,18 +391,15 @@ enum UpdateFulfillFetch {
 }
 
 /// The return type of get_update_fulfill_htlc_and_commit.
-pub enum UpdateFulfillCommitFetch {
+pub enum UpdateFulfillCommitFetch<'a> {
 	/// Indicates the HTLC fulfill is new, and either generated an update_fulfill message, placed
 	/// it in the holding cell, or re-generated the update_fulfill message after the same claim was
 	/// previously placed in the holding cell (and has since been removed).
 	NewClaim {
 		/// The ChannelMonitorUpdate which places the new payment preimage in the channel monitor
-		monitor_update: ChannelMonitorUpdate,
+		monitor_update: &'a ChannelMonitorUpdate,
 		/// The value of the HTLC which was claimed, in msat.
 		htlc_value_msat: u64,
-		/// The update_fulfill message and commitment_signed message (if the claim was not placed
-		/// in the holding cell).
-		msgs: Option<(msgs::UpdateFulfillHTLC, msgs::CommitmentSigned)>,
 	},
 	/// Indicates the HTLC fulfill is duplicative and already existed either in the holding cell
 	/// or has been forgotten (presumably previously claimed).
@@ -1922,22 +1919,30 @@ impl<Signer: Sign> Channel<Signer> {
 		}
 	}
 
-	pub fn get_update_fulfill_htlc_and_commit<L: Deref>(&mut self, htlc_id: u64, payment_preimage: PaymentPreimage, logger: &L) -> Result<UpdateFulfillCommitFetch, (ChannelError, ChannelMonitorUpdate)> where L::Target: Logger {
+	pub fn get_update_fulfill_htlc_and_commit<L: Deref>(&mut self, htlc_id: u64, payment_preimage: PaymentPreimage, logger: &L) -> UpdateFulfillCommitFetch where L::Target: Logger {
 		match self.get_update_fulfill_htlc(htlc_id, payment_preimage, logger) {
-			UpdateFulfillFetch::NewClaim { mut monitor_update, htlc_value_msat, msg: Some(update_fulfill_htlc) } => {
-				let (commitment, mut additional_update) = match self.send_commitment_no_status_check(logger) {
-					Err(e) => return Err((e, monitor_update)),
-					Ok(res) => res
-				};
-				// send_commitment_no_status_check may bump latest_monitor_id but we want them to be
+			UpdateFulfillFetch::NewClaim { mut monitor_update, htlc_value_msat, msg: Some(_) } => {
+				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);
-				Ok(UpdateFulfillCommitFetch::NewClaim { monitor_update, htlc_value_msat, msgs: Some((update_fulfill_htlc, commitment)) })
+				self.monitor_updating_paused(false, true, false, Vec::new(), Vec::new(), Vec::new());
+				self.pending_monitor_updates.push(monitor_update);
+				UpdateFulfillCommitFetch::NewClaim {
+					monitor_update: self.pending_monitor_updates.last().unwrap(),
+					htlc_value_msat,
+				}
 			},
-			UpdateFulfillFetch::NewClaim { monitor_update, htlc_value_msat, msg: None } =>
-				Ok(UpdateFulfillCommitFetch::NewClaim { monitor_update, htlc_value_msat, msgs: None }),
-			UpdateFulfillFetch::DuplicateClaim {} => Ok(UpdateFulfillCommitFetch::DuplicateClaim {}),
+			UpdateFulfillFetch::NewClaim { monitor_update, htlc_value_msat, msg: None } => {
+				self.monitor_updating_paused(false, false, false, Vec::new(), Vec::new(), Vec::new());
+				self.pending_monitor_updates.push(monitor_update);
+				UpdateFulfillCommitFetch::NewClaim {
+					monitor_update: self.pending_monitor_updates.last().unwrap(),
+					htlc_value_msat,
+				}
+			}
+			UpdateFulfillFetch::DuplicateClaim {} => UpdateFulfillCommitFetch::DuplicateClaim {},
 		}
 	}
 

lightning/src/ln/channelmanager.rs

@@ -4344,60 +4344,35 @@ impl<M: Deref, T: Deref, K: Deref, F: Deref, L: Deref> ChannelManager<M, T, K, F
 		let chan_id = prev_hop.outpoint.to_channel_id();
 		let channel_state = &mut *channel_state_lock;
 		if let hash_map::Entry::Occupied(mut chan) = channel_state.by_id.entry(chan_id) {
-			match chan.get_mut().get_update_fulfill_htlc_and_commit(prev_hop.htlc_id, payment_preimage, &self.logger) {
-				Ok(msgs_monitor_option) => {
-					if let UpdateFulfillCommitFetch::NewClaim { msgs, htlc_value_msat, monitor_update } = msgs_monitor_option {
-						match self.chain_monitor.update_channel(chan.get().get_funding_txo().unwrap(), &monitor_update) {
-							ChannelMonitorUpdateStatus::Completed => {},
-							e => {
-								log_given_level!(self.logger, if e == ChannelMonitorUpdateStatus::PermanentFailure { Level::Error } else { Level::Debug },
-									"Failed to update channel monitor with preimage {:?}: {:?}",
-									payment_preimage, e);
-								self.handle_monitor_update_completion_actions(completion_action(Some(htlc_value_msat)));
-								return Err((
-									chan.get().get_counterparty_node_id(),
-									handle_monitor_update_res!(self, e, chan, RAACommitmentOrder::CommitmentFirst, false, msgs.is_some()).unwrap_err(),
-								));
-							}
-						}
-						if let Some((msg, commitment_signed)) = msgs {
-							log_debug!(self.logger, "Claiming funds for HTLC with preimage {} resulted in a commitment_signed for channel {}",
-								log_bytes!(payment_preimage.0), log_bytes!(chan.get().channel_id()));
-							channel_state.pending_msg_events.push(events::MessageSendEvent::UpdateHTLCs {
-								node_id: chan.get().get_counterparty_node_id(),
-								updates: msgs::CommitmentUpdate {
-									update_add_htlcs: Vec::new(),
-									update_fulfill_htlcs: vec![msg],
-									update_fail_htlcs: Vec::new(),
-									update_fail_malformed_htlcs: Vec::new(),
-									update_fee: None,
-									commitment_signed,
-								}
-							});
-						}
-						self.handle_monitor_update_completion_actions(completion_action(Some(htlc_value_msat)));
-						Ok(())
-					} else {
-						Ok(())
-					}
-				},
-				Err((e, monitor_update)) => {
-					match self.chain_monitor.update_channel(chan.get().get_funding_txo().unwrap(), &monitor_update) {
-						ChannelMonitorUpdateStatus::Completed => {},
-						e => {
-							log_given_level!(self.logger, if e == ChannelMonitorUpdateStatus::PermanentFailure { Level::Error } else { Level::Info },
-								"Failed to update channel monitor with preimage {:?} immediately prior to force-close: {:?}",
-								payment_preimage, e);
-						},
-					}
-					let counterparty_node_id = chan.get().get_counterparty_node_id();
-					let (drop, res) = convert_chan_err!(self, e, chan.get_mut(), &chan_id);
-					if drop {
-						chan.remove_entry();
-					}
-					self.handle_monitor_update_completion_actions(completion_action(None));
-					Err((counterparty_node_id, res))
-				},
+			let counterparty_node_id = chan.get().get_counterparty_node_id();
+			let push_action = |htlc_value_msat| {
+				if let Some(action) = completion_action(htlc_value_msat) {
+					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();
+					log_trace!(self.logger, "Tracking monitor update completion action for channel {}: {:?}",
+						log_bytes!(chan_id), action);
+					peer_state.monitor_update_blocked_actions.entry(chan_id).or_insert(Vec::new()).push(action);
+				}
+			};
+
+			let monitor_option = chan.get_mut().get_update_fulfill_htlc_and_commit(prev_hop.htlc_id, payment_preimage, &self.logger);
+			if let UpdateFulfillCommitFetch::NewClaim { htlc_value_msat, monitor_update } = monitor_option {
+				push_action(Some(htlc_value_msat));
+				let update_id = monitor_update.update_id;
+				let update_res = self.chain_monitor.update_channel(prev_hop.outpoint, monitor_update);
+				let res = handle_new_monitor_update!(self, update_res, update_id, channel_state_lock, channel_state.pending_msg_events, chan);
+				if let Err(e) = res {
+					// TODO: This is a *critical* error - we probably updated some other
+					// monitors with a preimage. We should retry this monitor udpate over
+					// and over again until morale improves.
+					log_error!(self.logger, "Failed to update channel monitor with preimage {:?}", payment_preimage);
+					return Err((counterparty_node_id, e));
+				}
+				Ok(())
+			} else {
+				Ok(())
 			}
 		} else {
 			let preimage_update = ChannelMonitorUpdate {
@@ -7229,8 +7204,6 @@ impl<M: Deref, T: Deref, K: Deref, F: Deref, L: Deref> Writeable for ChannelMana
 			// LDK versions prior to 0.0.113 do not know how to read the pending claimed payments
 			// map. Thus, if there are no entries we skip writing a TLV for it.
 			pending_claimed_payments = None;
-		} else {
-			debug_assert!(false, "While we have code to serialize pending_claimed_payments, the map should always be empty until a later PR");
 		}
 
 		write_tlv_fields!(writer, {