Delay RAA-after-next processing until PaymentSent is are handled

In 0ad1f4c we fixed a nasty bug
where a failure to persist a `ChannelManager` faster than a
`ChannelMonitor` could result in the loss of a `PaymentSent` event,
eventually resulting in a `PaymentFailed` instead!

As noted in that commit, there's still some risk, though its been
substantially reduced - if we receive an `update_fulfill_htlc`
message for an outbound payment, and persist the initial removal
`ChannelMonitorUpdate`, then respond with our own
`commitment_signed` + `revoke_and_ack`, followed by receiving our
peer's final `revoke_and_ack`, and then persist the
`ChannelMonitorUpdate` generated from that, all prior to completing
a `ChannelManager` persistence, we'll still forget the HTLC and
eventually trigger a `PaymentFailed` rather than the correct
`PaymentSent`.

Here we fully fix the issue by delaying the final
`ChannelMonitorUpdate` persistence until the `PaymentSent` event
has been processed and document the fact that a spurious
`PaymentFailed` event can still be generated for a sent payment.

The original fix in 0ad1f4c is
still incredibly useful here, allowing us to avoid blocking the
first `ChannelMonitorUpdate` until the event processing completes,
as this would cause us to add event-processing delay in our general
commitment update latency. Instead, we ultimately race the user
handling the `PaymentSent` event with how long it takes our
`revoke_and_ack` + `commitment_signed` to make it to our
counterparty and receive the response `revoke_and_ack`. This should
give the user plenty of time to handle the event before we need to
make progress.

Sadly, because we change our `ChannelMonitorUpdate` semantics, this
change requires a number of test changes, avoiding checking for a
post-RAA `ChannelMonitorUpdate` until after we process a
`PaymentSent` event. Note that this does not apply to payments we
learned the preimage for on-chain - ensuring `PaymentSent` events
from such resolutions will be addressed in a future PR. Thus, tests
which resolve payments on-chain switch to a direct call to the
`expect_payment_sent` function with the claim-expected flag unset.

Author: TheBlueMatt

lightning-invoice/src/utils.rs

@@ -1364,22 +1364,7 @@ mod test {
 		let payment_preimage_opt = if user_generated_pmt_hash { None } else { Some(payment_preimage) };
 		expect_payment_claimable!(&nodes[fwd_idx], payment_hash, payment_secret, payment_amt, payment_preimage_opt, invoice.recover_payee_pub_key());
 		do_claim_payment_along_route(&nodes[0], &[&vec!(&nodes[fwd_idx])[..]], false, payment_preimage);
-		let events = nodes[0].node.get_and_clear_pending_events();
-		assert_eq!(events.len(), 2);
-		match events[0] {
-			Event::PaymentSent { payment_preimage: ref ev_preimage, payment_hash: ref ev_hash, ref fee_paid_msat, .. } => {
-				assert_eq!(payment_preimage, *ev_preimage);
-				assert_eq!(payment_hash, *ev_hash);
-				assert_eq!(fee_paid_msat, &Some(0));
-			},
-			_ => panic!("Unexpected event")
-		}
-		match events[1] {
-			Event::PaymentPathSuccessful { payment_hash: hash, .. } => {
-				assert_eq!(hash, Some(payment_hash));
-			},
-			_ => panic!("Unexpected event")
-		}
+		expect_payment_sent(&nodes[0], payment_preimage, None, true, true);
 	}
 
 	#[test]

lightning/src/chain/chainmonitor.rs

@@ -826,7 +826,7 @@ impl<ChannelSigner: WriteableEcdsaChannelSigner, C: Deref, T: Deref, F: Deref, L
 #[cfg(test)]
 mod tests {
 	use crate::{check_added_monitors, check_closed_broadcast, check_closed_event};
-	use crate::{expect_payment_sent, expect_payment_claimed, expect_payment_sent_without_paths, expect_payment_path_successful, get_event_msg};
+	use crate::{expect_payment_claimed, expect_payment_path_successful, get_event_msg};
 	use crate::{get_htlc_update_msgs, get_local_commitment_txn, get_revoke_commit_msgs, get_route_and_payment_hash, unwrap_send_err};
 	use crate::chain::{ChannelMonitorUpdateStatus, Confirm, Watch};
 	use crate::chain::channelmonitor::LATENCY_GRACE_PERIOD_BLOCKS;
@@ -909,7 +909,7 @@ mod tests {
 
 		let updates = get_htlc_update_msgs!(nodes[1], nodes[0].node.get_our_node_id());
 		nodes[0].node.handle_update_fulfill_htlc(&nodes[1].node.get_our_node_id(), &updates.update_fulfill_htlcs[0]);
-		expect_payment_sent_without_paths!(nodes[0], payment_preimage_1);
+		expect_payment_sent(&nodes[0], payment_preimage_1, None, false, false);
 		nodes[0].node.handle_commitment_signed(&nodes[1].node.get_our_node_id(), &updates.commitment_signed);
 		check_added_monitors!(nodes[0], 1);
 		let (as_first_raa, as_first_update) = get_revoke_commit_msgs!(nodes[0], nodes[1].node.get_our_node_id());
@@ -922,7 +922,7 @@ mod tests {
 		let bs_first_raa = get_event_msg!(nodes[1], MessageSendEvent::SendRevokeAndACK, nodes[0].node.get_our_node_id());
 
 		nodes[0].node.handle_update_fulfill_htlc(&nodes[1].node.get_our_node_id(), &bs_second_updates.update_fulfill_htlcs[0]);
-		expect_payment_sent_without_paths!(nodes[0], payment_preimage_2);
+		expect_payment_sent(&nodes[0], payment_preimage_2, None, false, false);
 		nodes[0].node.handle_commitment_signed(&nodes[1].node.get_our_node_id(), &bs_second_updates.commitment_signed);
 		check_added_monitors!(nodes[0], 1);
 		nodes[0].node.handle_revoke_and_ack(&nodes[1].node.get_our_node_id(), &bs_first_raa);
@@ -1005,7 +1005,7 @@ mod tests {
 			}
 		}
 
-		expect_payment_sent!(nodes[0], payment_preimage);
+		expect_payment_sent(&nodes[0], payment_preimage, None, true, false);
 	}
 
 	#[test]

lightning/src/events/mod.rs

@@ -472,6 +472,11 @@ pub enum Event {
 	/// payment is no longer retryable, due either to the [`Retry`] provided or
 	/// [`ChannelManager::abandon_payment`] having been called for the corresponding payment.
 	///
+	/// In exceedingly rare cases, it is possible that an [`Event::PaymentFailed`] is generated for
+	/// a payment after an [`Event::PaymentSent`] event for this same payment has already been
+	/// received and processed. In this case, the [`Event::PaymentFailed`] event MUST be ignored,
+	/// and the payment MUST be treated as having succeeded.
+	///
 	/// [`Retry`]: crate::ln::channelmanager::Retry
 	/// [`ChannelManager::abandon_payment`]: crate::ln::channelmanager::ChannelManager::abandon_payment
 	PaymentFailed {

lightning/src/ln/chanmon_update_fail_tests.rs

@@ -1400,6 +1400,7 @@ fn claim_while_disconnected_monitor_update_fail() {
 		MessageSendEvent::UpdateHTLCs { ref node_id, ref updates } => {
 			assert_eq!(*node_id, nodes[0].node.get_our_node_id());
 			nodes[0].node.handle_update_fulfill_htlc(&nodes[1].node.get_our_node_id(), &updates.update_fulfill_htlcs[0]);
+			expect_payment_sent(&nodes[0], payment_preimage_1, None, false, false);
 			nodes[0].node.handle_commitment_signed(&nodes[1].node.get_our_node_id(), &updates.commitment_signed);
 			check_added_monitors!(nodes[0], 1);
 
@@ -1437,7 +1438,7 @@ fn claim_while_disconnected_monitor_update_fail() {
 
 	nodes[0].node.handle_revoke_and_ack(&nodes[1].node.get_our_node_id(), &bs_raa);
 	check_added_monitors!(nodes[0], 1);
-	expect_payment_sent!(nodes[0], payment_preimage_1);
+	expect_payment_path_successful!(nodes[0]);
 
 	claim_payment(&nodes[0], &[&nodes[1]], payment_preimage_2);
 }
@@ -2191,7 +2192,7 @@ fn test_fail_htlc_on_broadcast_after_claim() {
 	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_id_2 }]);
 
 	nodes[0].node.handle_update_fulfill_htlc(&nodes[1].node.get_our_node_id(), &bs_updates.update_fulfill_htlcs[0]);
-	expect_payment_sent_without_paths!(nodes[0], payment_preimage);
+	expect_payment_sent(&nodes[0], payment_preimage, None, false, false);
 	commitment_signed_dance!(nodes[0], nodes[1], bs_updates.commitment_signed, true, true);
 	expect_payment_path_successful!(nodes[0]);
 }
@@ -2437,7 +2438,7 @@ fn do_channel_holding_cell_serialize(disconnect: bool, reload_a: bool) {
 			assert!(updates.update_fee.is_none());
 			assert_eq!(updates.update_fulfill_htlcs.len(), 1);
 			nodes[1].node.handle_update_fulfill_htlc(&nodes[0].node.get_our_node_id(), &updates.update_fulfill_htlcs[0]);
-			expect_payment_sent_without_paths!(nodes[1], payment_preimage_0);
+			expect_payment_sent(&nodes[1], payment_preimage_0, None, false, false);
 			assert_eq!(updates.update_add_htlcs.len(), 1);
 			nodes[1].node.handle_update_add_htlc(&nodes[0].node.get_our_node_id(), &updates.update_add_htlcs[0]);
 			updates.commitment_signed
@@ -2454,7 +2455,7 @@ fn do_channel_holding_cell_serialize(disconnect: bool, reload_a: bool) {
 	expect_payment_claimable!(nodes[1], payment_hash_1, payment_secret_1, 100000);
 	check_added_monitors!(nodes[1], 1);
 
-	commitment_signed_dance!(nodes[1], nodes[0], (), false, true, false);
+	commitment_signed_dance!(nodes[1], nodes[0], (), false, true, false, false);
 
 	let events = nodes[1].node.get_and_clear_pending_events();
 	assert_eq!(events.len(), 2);
@@ -2555,7 +2556,7 @@ fn do_test_reconnect_dup_htlc_claims(htlc_status: HTLCStatusAtDupClaim, second_f
 		bs_updates = Some(get_htlc_update_msgs!(nodes[1], nodes[0].node.get_our_node_id()));
 		assert_eq!(bs_updates.as_ref().unwrap().update_fulfill_htlcs.len(), 1);
 		nodes[0].node.handle_update_fulfill_htlc(&nodes[1].node.get_our_node_id(), &bs_updates.as_ref().unwrap().update_fulfill_htlcs[0]);
-		expect_payment_sent_without_paths!(nodes[0], payment_preimage);
+		expect_payment_sent(&nodes[0], payment_preimage, None, false, false);
 		if htlc_status == HTLCStatusAtDupClaim::Cleared {
 			commitment_signed_dance!(nodes[0], nodes[1], &bs_updates.as_ref().unwrap().commitment_signed, false);
 			expect_payment_path_successful!(nodes[0]);
@@ -2582,7 +2583,7 @@ fn do_test_reconnect_dup_htlc_claims(htlc_status: HTLCStatusAtDupClaim, second_f
 		bs_updates = Some(get_htlc_update_msgs!(nodes[1], nodes[0].node.get_our_node_id()));
 		assert_eq!(bs_updates.as_ref().unwrap().update_fulfill_htlcs.len(), 1);
 		nodes[0].node.handle_update_fulfill_htlc(&nodes[1].node.get_our_node_id(), &bs_updates.as_ref().unwrap().update_fulfill_htlcs[0]);
-		expect_payment_sent_without_paths!(nodes[0], payment_preimage);
+		expect_payment_sent(&nodes[0], payment_preimage, None, false, false);
 	}
 	if htlc_status != HTLCStatusAtDupClaim::Cleared {
 		commitment_signed_dance!(nodes[0], nodes[1], &bs_updates.as_ref().unwrap().commitment_signed, false);
@@ -2779,7 +2780,7 @@ fn double_temp_error() {
 	assert_eq!(node_id, nodes[0].node.get_our_node_id());
 	nodes[0].node.handle_update_fulfill_htlc(&nodes[1].node.get_our_node_id(), &update_fulfill_1);
 	check_added_monitors!(nodes[0], 0);
-	expect_payment_sent_without_paths!(nodes[0], payment_preimage_1);
+	expect_payment_sent(&nodes[0], payment_preimage_1, None, false, false);
 	nodes[0].node.handle_commitment_signed(&nodes[1].node.get_our_node_id(), &commitment_signed_b1);
 	check_added_monitors!(nodes[0], 1);
 	nodes[0].node.process_pending_htlc_forwards();
@@ -3004,3 +3005,67 @@ fn test_inbound_reload_without_init_mon() {
 	do_test_inbound_reload_without_init_mon(false, true);
 	do_test_inbound_reload_without_init_mon(false, false);
 }
+
+#[test]
+fn test_blocked_chan_preimage_release() {
+	// Test that even if a channel's `ChannelMonitorUpdate` flow is blocked waiting on an event to
+	// be handled HTLC preimage `ChannelMonitorUpdate`s will still go out.
+	let chanmon_cfgs = create_chanmon_cfgs(3);
+	let node_cfgs = create_node_cfgs(3, &chanmon_cfgs);
+	let node_chanmgrs = create_node_chanmgrs(3, &node_cfgs, &[None, None, None]);
+	let mut nodes = create_network(3, &node_cfgs, &node_chanmgrs);
+
+	create_announced_chan_between_nodes(&nodes, 0, 1).2;
+	create_announced_chan_between_nodes(&nodes, 1, 2).2;
+
+	send_payment(&nodes[0], &[&nodes[1], &nodes[2]], 5_000_000);
+
+	// Tee up two payments in opposite directions across nodes[1], one it sent to generate a
+	// PaymentSent event and one it forwards.
+	let (payment_preimage_1, payment_hash_1, _) = route_payment(&nodes[1], &[&nodes[2]], 1_000_000);
+	let (payment_preimage_2, payment_hash_2, _) = route_payment(&nodes[2], &[&nodes[1], &nodes[0]], 1_000_000);
+
+	// Claim the first payment to get a `PaymentSent` event (but don't handle it yet).
+	nodes[2].node.claim_funds(payment_preimage_1);
+	check_added_monitors(&nodes[2], 1);
+	expect_payment_claimed!(nodes[2], payment_hash_1, 1_000_000);
+
+	let cs_htlc_fulfill_updates = get_htlc_update_msgs!(nodes[2], nodes[1].node.get_our_node_id());
+	nodes[1].node.handle_update_fulfill_htlc(&nodes[2].node.get_our_node_id(), &cs_htlc_fulfill_updates.update_fulfill_htlcs[0]);
+	commitment_signed_dance!(nodes[1], nodes[2], cs_htlc_fulfill_updates.commitment_signed, false);
+	check_added_monitors(&nodes[1], 0);
+
+	// Now claim the second payment on nodes[0], which will ultimately result in nodes[1] trying to
+	// claim an HTLC on its channel with nodes[2], but that channel is blocked on the above
+	// `PaymentSent` event.
+	nodes[0].node.claim_funds(payment_preimage_2);
+	check_added_monitors(&nodes[0], 1);
+	expect_payment_claimed!(nodes[0], payment_hash_2, 1_000_000);
+
+	let as_htlc_fulfill_updates = get_htlc_update_msgs!(nodes[0], nodes[1].node.get_our_node_id());
+	nodes[1].node.handle_update_fulfill_htlc(&nodes[0].node.get_our_node_id(), &as_htlc_fulfill_updates.update_fulfill_htlcs[0]);
+	check_added_monitors(&nodes[1], 1); // We generate only a preimage monitor update
+	assert!(nodes[1].node.get_and_clear_pending_msg_events().is_empty());
+
+	// Finish the CS dance between nodes[0] and nodes[1].
+	commitment_signed_dance!(nodes[1], nodes[0], as_htlc_fulfill_updates.commitment_signed, false);
+	check_added_monitors(&nodes[1], 0);
+
+	let events = nodes[1].node.get_and_clear_pending_events();
+	assert_eq!(events.len(), 3);
+	if let Event::PaymentSent { .. } = events[0] {} else { panic!(); }
+	if let Event::PaymentPathSuccessful { .. } = events[2] {} else { panic!(); }
+	if let Event::PaymentForwarded { .. } = events[1] {} else { panic!(); }
+
+	// The event processing should release the last RAA update.
+	check_added_monitors(&nodes[1], 1);
+
+	// When we fetch the next update the message getter will generate the next update for nodes[2],
+	// generating a further monitor update.
+	let bs_htlc_fulfill_updates = get_htlc_update_msgs!(nodes[1], nodes[2].node.get_our_node_id());
+	check_added_monitors(&nodes[1], 1);
+
+	nodes[2].node.handle_update_fulfill_htlc(&nodes[1].node.get_our_node_id(), &bs_htlc_fulfill_updates.update_fulfill_htlcs[0]);
+	commitment_signed_dance!(nodes[2], nodes[1], bs_htlc_fulfill_updates.commitment_signed, false);
+	expect_payment_sent(&nodes[2], payment_preimage_2, None, true, true);
+}

lightning/src/ln/channel.rs

@@ -3143,7 +3143,8 @@ 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<(Vec<(HTLCSource, PaymentHash)>, Option<&ChannelMonitorUpdate>), ChannelError>
+	pub fn revoke_and_ack<L: Deref>(&mut self, msg: &msgs::RevokeAndACK, logger: &L, hold_mon_update: bool)
+	-> Result<(Vec<(HTLCSource, PaymentHash)>, Option<&ChannelMonitorUpdate>), ChannelError>
 		where L::Target: Logger,
 	{
 		if (self.context.channel_state & (ChannelState::ChannelReady as u32)) != (ChannelState::ChannelReady as u32) {
@@ -3323,6 +3324,10 @@ impl<Signer: WriteableEcdsaChannelSigner> Channel<Signer> {
 			}
 		}
 
+		let release_monitor = self.context.pending_monitor_updates.iter().all(|upd| !upd.blocked) && !hold_mon_update;
+		let release_state_str =
+			if hold_mon_update { "Holding" } else if release_monitor { "Releasing" } else { "Blocked" };
+
 		if (self.context.channel_state & ChannelState::MonitorUpdateInProgress as u32) == ChannelState::MonitorUpdateInProgress as u32 {
 			// We can't actually generate a new commitment transaction (incl by freeing holding
 			// cells) while we can't update the monitor, so we just return what we have.
@@ -3341,7 +3346,11 @@ impl<Signer: WriteableEcdsaChannelSigner> Channel<Signer> {
 			self.context.monitor_pending_failures.append(&mut revoked_htlcs);
 			self.context.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.context.channel_id()));
-			return Ok((Vec::new(), self.push_ret_blockable_mon_update(monitor_update)));
+			self.context.pending_monitor_updates.push(PendingChannelMonitorUpdate {
+				update: monitor_update, blocked: !release_monitor,
+			});
+			return Ok((Vec::new(),
+				if release_monitor { self.context.pending_monitor_updates.last().map(|upd| &upd.update) } else { None }));
 		}
 
 		match self.free_holding_cell_htlcs(logger) {
@@ -3352,8 +3361,15 @@ impl<Signer: WriteableEcdsaChannelSigner> Channel<Signer> {
 				self.context.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 {} with holding cell HTLCs freed. {} monitor update.",
+					log_bytes!(self.context.channel_id()), release_state_str);
+
 				self.monitor_updating_paused(false, true, false, to_forward_infos, revoked_htlcs, finalized_claimed_htlcs);
-				Ok((htlcs_to_fail, self.push_ret_blockable_mon_update(monitor_update)))
+				self.context.pending_monitor_updates.push(PendingChannelMonitorUpdate {
+					update: monitor_update, blocked: !release_monitor,
+				});
+				Ok((htlcs_to_fail,
+					if release_monitor { self.context.pending_monitor_updates.last().map(|upd| &upd.update) } else { None }))
 			},
 			(None, htlcs_to_fail) => {
 				if require_commitment {
@@ -3364,14 +3380,27 @@ impl<Signer: WriteableEcdsaChannelSigner> Channel<Signer> {
 					self.context.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.context.channel_id()), update_fail_htlcs.len() + update_fail_malformed_htlcs.len());
+					log_debug!(logger, "Received a valid revoke_and_ack for channel {}. Responding with a commitment update with {} HTLCs failed. {} monitor update.",
+						log_bytes!(self.context.channel_id()),
+						update_fail_htlcs.len() + update_fail_malformed_htlcs.len(),
+						release_state_str);
+
 					self.monitor_updating_paused(false, true, false, to_forward_infos, revoked_htlcs, finalized_claimed_htlcs);
-					Ok((htlcs_to_fail, self.push_ret_blockable_mon_update(monitor_update)))
+					self.context.pending_monitor_updates.push(PendingChannelMonitorUpdate {
+						update: monitor_update, blocked: !release_monitor,
+					});
+					Ok((htlcs_to_fail,
+						if release_monitor { self.context.pending_monitor_updates.last().map(|upd| &upd.update) } else { None }))
 				} else {
-					log_debug!(logger, "Received a valid revoke_and_ack for channel {} with no reply necessary.", log_bytes!(self.context.channel_id()));
+					log_debug!(logger, "Received a valid revoke_and_ack for channel {} with no reply necessary. {} monitor update.",
+						log_bytes!(self.context.channel_id()), release_state_str);
+
 					self.monitor_updating_paused(false, false, false, to_forward_infos, revoked_htlcs, finalized_claimed_htlcs);
-					Ok((htlcs_to_fail, self.push_ret_blockable_mon_update(monitor_update)))
+					self.context.pending_monitor_updates.push(PendingChannelMonitorUpdate {
+						update: monitor_update, blocked: !release_monitor,
+					});
+					Ok((htlcs_to_fail,
+						if release_monitor { self.context.pending_monitor_updates.last().map(|upd| &upd.update) } else { None }))
 				}
 			}
 		}