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

@@ -1091,22 +1091,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, route.paths[0].last().unwrap().pubkey);
 		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

@@ -788,7 +788,7 @@ mod tests {
 	use bitcoin::{BlockHeader, TxMerkleNode};
 	use bitcoin::hashes::Hash;
 	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;
@@ -871,7 +871,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());
@@ -884,7 +884,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);
@@ -972,7 +972,7 @@ mod tests {
 			}
 		}
 
-		expect_payment_sent!(nodes[0], payment_preimage);
+		expect_payment_sent(&nodes[0], payment_preimage, None, true, false);
 	}
 
 	#[test]

lightning/src/ln/chanmon_update_fail_tests.rs

@@ -1371,6 +1371,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);
 
@@ -1408,7 +1409,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);
 }
@@ -2140,7 +2141,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]);
 }
@@ -2376,7 +2377,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
@@ -2393,7 +2394,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);
@@ -2493,7 +2494,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]);
@@ -2520,7 +2521,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);
@@ -2717,7 +2718,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();
@@ -2940,3 +2941,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 let the last RAA update fly.
+	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

@@ -3427,7 +3427,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.channel_state & (ChannelState::ChannelReady as u32)) != (ChannelState::ChannelReady as u32) {
@@ -3624,7 +3625,7 @@ 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()));
-			let fly_monitor = self.pending_monitor_updates.iter().all(|upd| upd.flown);
+			let fly_monitor = self.pending_monitor_updates.iter().all(|upd| upd.flown) && !hold_mon_update;
 			self.pending_monitor_updates.push(PendingChannelMonitorUpdate {
 				update: monitor_update, flown: fly_monitor,
 			});
@@ -3641,7 +3642,7 @@ impl<Signer: WriteableEcdsaChannelSigner> Channel<Signer> {
 				monitor_update.updates.append(&mut additional_update.updates);
 
 				self.monitor_updating_paused(false, true, false, to_forward_infos, revoked_htlcs, finalized_claimed_htlcs);
-				let fly_monitor = self.pending_monitor_updates.iter().all(|upd| upd.flown);
+				let fly_monitor = self.pending_monitor_updates.iter().all(|upd| upd.flown) && !hold_mon_update;
 				self.pending_monitor_updates.push(PendingChannelMonitorUpdate {
 					update: monitor_update, flown: fly_monitor,
 				});
@@ -3660,7 +3661,7 @@ impl<Signer: WriteableEcdsaChannelSigner> Channel<Signer> {
 					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());
 					self.monitor_updating_paused(false, true, false, to_forward_infos, revoked_htlcs, finalized_claimed_htlcs);
-					let fly_monitor = self.pending_monitor_updates.iter().all(|upd| upd.flown);
+					let fly_monitor = self.pending_monitor_updates.iter().all(|upd| upd.flown) && !hold_mon_update;
 					self.pending_monitor_updates.push(PendingChannelMonitorUpdate {
 						update: monitor_update, flown: fly_monitor,
 					});
@@ -3669,7 +3670,7 @@ impl<Signer: WriteableEcdsaChannelSigner> Channel<Signer> {
 				} else {
 					log_debug!(logger, "Received a valid revoke_and_ack for channel {} with no reply necessary.", log_bytes!(self.channel_id()));
 					self.monitor_updating_paused(false, false, false, to_forward_infos, revoked_htlcs, finalized_claimed_htlcs);
-					let fly_monitor = self.pending_monitor_updates.iter().all(|upd| upd.flown);
+					let fly_monitor = self.pending_monitor_updates.iter().all(|upd| upd.flown) && !hold_mon_update;
 					self.pending_monitor_updates.push(PendingChannelMonitorUpdate {
 						update: monitor_update, flown: fly_monitor,
 					});