Only mark all mon updates complete if there are no blocked updates

In `handle_new_monitor_update!`, we correctly check that the
channel doesn't have any blocked monitor updates pending before
calling `handle_monitor_update_completion!` (which calls
`Channel::monitor_updating_restored`, which in turn assumes that
all generated `ChannelMonitorUpdate`s, including blocked ones, have
completed).

We, however, did not do the same check at several other places
where we called `handle_monitor_update_completion!`. Specifically,
after a monitor update completes during reload (processed via a
`BackgroundEvent` or when monitor update completes async, we didn't
check if there were any blocked monitor updates before completing).

Here we add the missing check, as well as an assertion in
`Channel::monitor_updating_restored`.

Conflicts resolved in:
 * lightning/src/ln/chanmon_update_fail_tests.rs due to
   `rustfmt`-induced changes as well as other tests cleanups.
 * lightning/src/ln/channelmanager.rs due to upstream Channel object
   refactoring

Author: TheBlueMatt

lightning/src/ln/chanmon_update_fail_tests.rs

@@ -2931,16 +2931,28 @@ fn test_inbound_reload_without_init_mon() {
 	do_test_inbound_reload_without_init_mon(false, false);
 }
 
-#[test]
-fn test_blocked_chan_preimage_release() {
+#[derive(PartialEq, Eq)]
+enum BlockedUpdateComplMode {
+	Async,
+	AtReload,
+	Sync,
+}
+
+fn do_test_blocked_chan_preimage_release(completion_mode: BlockedUpdateComplMode) {
 	// 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 persister;
+	let new_chain_mon;
 	let node_chanmgrs = create_node_chanmgrs(3, &node_cfgs, &[None, None, None]);
+	let nodes_1_reload;
 	let mut nodes = create_network(3, &node_cfgs, &node_chanmgrs);
 
-	create_announced_chan_between_nodes(&nodes, 0, 1);
+	let node_a_id = nodes[0].node.get_our_node_id();
+	let node_b_id = nodes[1].node.get_our_node_id();
+
+	let chan_id_1 = create_announced_chan_between_nodes(&nodes, 0, 1).2;
 	let chan_id_2 = create_announced_chan_between_nodes(&nodes, 1, 2).2;
 
 	send_payment(&nodes[0], &[&nodes[1], &nodes[2]], 5_000_000);
@@ -2968,25 +2980,62 @@ fn test_blocked_chan_preimage_release() {
 	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());
+	if completion_mode != BlockedUpdateComplMode::Sync {
+		// We use to incorrectly handle monitor update completion in cases where we completed a
+		// monitor update async or after reload. We test both based on the `completion_mode`.
+		chanmon_cfgs[1].persister.set_update_ret(ChannelMonitorUpdateStatus::InProgress);
+	}
 	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!(get_monitor!(nodes[1], chan_id_2).get_stored_preimages().contains_key(&payment_hash_2));
 	assert!(nodes[1].node.get_and_clear_pending_msg_events().is_empty());
+	if completion_mode == BlockedUpdateComplMode::AtReload {
+		let node_ser = nodes[1].node.encode();
+		let chan_mon_0 = get_monitor!(nodes[1], chan_id_1).encode();
+		let chan_mon_1 = get_monitor!(nodes[1], chan_id_2).encode();
+
+		let mons = &[&chan_mon_0[..], &chan_mon_1[..]];
+		reload_node!(nodes[1], &node_ser, mons, persister, new_chain_mon, nodes_1_reload);
+
+		nodes[0].node.peer_disconnected(node_b_id);
+		nodes[2].node.peer_disconnected(node_b_id);
+
+		let mut a_b_reconnect = ReconnectArgs::new(&nodes[0], &nodes[1]);
+		a_b_reconnect.pending_htlc_claims.1 = 1;
+		// Note that we will expect no final RAA monitor update in
+		// `commitment_signed_dance_through_cp_raa` during the reconnect, matching the below case.
+		reconnect_nodes(a_b_reconnect);
+		reconnect_nodes(ReconnectArgs::new(&nodes[2], &nodes[1]));
+	} else if completion_mode == BlockedUpdateComplMode::Async {
+		let (outpoint, latest_update, _) = nodes[1].chain_monitor.latest_monitor_update_id.lock().unwrap().get(&chan_id_2).unwrap().clone();
+		nodes[1]
+			.chain_monitor
+			.chain_monitor
+			.channel_monitor_updated(outpoint, latest_update)
+			.unwrap();
+	}
 
 	// Finish the CS dance between nodes[0] and nodes[1]. Note that until the event handling, the
 	// update_fulfill_htlc + CS is held, even though the preimage is already on disk for the
 	// channel.
-	nodes[1].node.handle_commitment_signed(nodes[0].node.get_our_node_id(), &as_htlc_fulfill_updates.commitment_signed);
-	check_added_monitors(&nodes[1], 1);
-	let (a, raa) = do_main_commitment_signed_dance(&nodes[1], &nodes[0], false);
-	assert!(a.is_none());
+	// Note that when completing as a side effect of a reload we completed the CS dance in
+	// `reconnect_nodes` above.
+	if completion_mode != BlockedUpdateComplMode::AtReload {
+		nodes[1].node.handle_commitment_signed(
+			node_a_id,
+			&as_htlc_fulfill_updates.commitment_signed,
+		);
+		check_added_monitors(&nodes[1], 1);
+		let (a, raa) = do_main_commitment_signed_dance(&nodes[1], &nodes[0], false);
+		assert!(a.is_none());
 
-	nodes[1].node.handle_revoke_and_ack(nodes[0].node.get_our_node_id(), &raa);
-	check_added_monitors(&nodes[1], 0);
-	assert!(nodes[1].node.get_and_clear_pending_msg_events().is_empty());
+		nodes[1].node.handle_revoke_and_ack(node_a_id, &raa);
+		check_added_monitors(&nodes[1], 0);
+		assert!(nodes[1].node.get_and_clear_pending_msg_events().is_empty());
+	}
 
 	let events = nodes[1].node.get_and_clear_pending_events();
-	assert_eq!(events.len(), 3);
+	assert_eq!(events.len(), 3, "{events:?}");
 	if let Event::PaymentSent { .. } = events[0] {} else { panic!(); }
 	if let Event::PaymentPathSuccessful { .. } = events[2] {} else { panic!(); }
 	if let Event::PaymentForwarded { .. } = events[1] {} else { panic!(); }
@@ -3004,6 +3053,13 @@ fn test_blocked_chan_preimage_release() {
 	expect_payment_sent(&nodes[2], payment_preimage_2, None, true, true);
 }
 
+#[test]
+fn test_blocked_chan_preimage_release() {
+	do_test_blocked_chan_preimage_release(BlockedUpdateComplMode::AtReload);
+	do_test_blocked_chan_preimage_release(BlockedUpdateComplMode::Sync);
+	do_test_blocked_chan_preimage_release(BlockedUpdateComplMode::Async);
+}
+
 fn do_test_inverted_mon_completion_order(with_latest_manager: bool, complete_bc_commitment_dance: bool) {
 	// When we forward a payment and receive `update_fulfill_htlc`+`commitment_signed` messages
 	// from the downstream channel, we immediately claim the HTLC on the upstream channel, before

lightning/src/ln/channel.rs

@@ -5941,6 +5941,7 @@ impl<SP: Deref> Channel<SP> where
 	{
 		assert!(self.context.channel_state.is_monitor_update_in_progress());
 		self.context.channel_state.clear_monitor_update_in_progress();
+		assert_eq!(self.blocked_monitor_updates_pending(), 0);
 
 		// If we're past (or at) the AwaitingChannelReady stage on an outbound channel, try to
 		// (re-)broadcast the funding transaction as we may have declined to broadcast it when we

lightning/src/ln/channelmanager.rs

@@ -6347,7 +6347,9 @@ where
 						let mut peer_state_lock = peer_state_mutex.lock().unwrap();
 						let peer_state = &mut *peer_state_lock;
 						if let Some(ChannelPhase::Funded(chan)) = peer_state.channel_by_id.get_mut(&channel_id) {
-							handle_monitor_update_completion!(self, peer_state_lock, peer_state, per_peer_state, chan);
+							if chan.blocked_monitor_updates_pending() == 0 {
+								handle_monitor_update_completion!(self, peer_state_lock, peer_state, per_peer_state, chan);
+							}
 						} else {
 							let update_actions = peer_state.monitor_update_blocked_actions
 								.remove(&channel_id).unwrap_or(Vec::new());
@@ -7625,8 +7627,12 @@ This indicates a bug inside LDK. Please report this error at https://github.com/
 
 		if let Some(ChannelPhase::Funded(chan)) = peer_state.channel_by_id.get_mut(channel_id) {
 			if chan.is_awaiting_monitor_update() {
-				log_trace!(logger, "Channel is open and awaiting update, resuming it");
-				handle_monitor_update_completion!(self, peer_state_lock, peer_state, per_peer_state, chan);
+				if chan.blocked_monitor_updates_pending() == 0 {
+					log_trace!(logger, "Channel is open and awaiting update, resuming it");
+					handle_monitor_update_completion!(self, peer_state_lock, peer_state, per_peer_state, chan);
+				} else {
+					log_trace!(logger, "Channel is open and awaiting update, leaving it blocked due to a blocked monitor update");
+				}
 			} else {
 				log_trace!(logger, "Channel is open but not awaiting update");
 			}