Hold in-flight monitor updates until background event processing

We previously assumed background events would eventually be processed
prior to another `ChannelManager` write, so we would immediately remove
all in-flight monitor updates that completed since the last
`ChannelManager` serialization. This isn't always the case, so we now
keep them all around until we're ready to handle them, i.e., when
`process_background_events` is called.

This was discovered while fuzzing `chanmon_consistency_target` on the
main branch with some changes that allow it to connect blocks. It was
triggered by reloading the `ChannelManager` after a monitor update
completion for an outgoing HTLC, calling
`ChannelManager::best_block_updated`, and reloading the `ChannelManager`
once again. A test is included that provides a minimal reproduction of
this case.

Author: wpaulino

lightning/src/ln/channelmanager.rs

@@ -8151,9 +8151,50 @@ impl<
 		for event in background_events.drain(..) {
 			match event {
 				BackgroundEvent::MonitorUpdateRegeneratedOnStartup { counterparty_node_id, funding_txo, channel_id, update } => {
+					// We previously assumed background events would eventually be processed prior
+					// to another `ChannelManager` write, so we would immediately remove all
+					// in-flight monitor updates that completed since the last `ChannelManager`
+					// serialization. This isn't always the case, so we now keep them all around
+					// until we're ready to handle them, i.e., when `process_background_events` is
+					// called.
+					//
+					// Remove the matching in-flight monitor update for this channel, as it has
+					// already been persisted to the monitor and can be applied to our internal
+					// state such that the channel may make forward progress.
+					{
+						let per_peer_state = self.per_peer_state.read().unwrap();
+						let mut peer_state_lock;
+						let peer_state_mutex_opt = per_peer_state.get(&counterparty_node_id);
+						if peer_state_mutex_opt.is_none() { continue; }
+						peer_state_lock = peer_state_mutex_opt.unwrap().lock().unwrap();
+						let peer_state = &mut *peer_state_lock;
+						if let Some((_, updates)) = peer_state.in_flight_monitor_updates.get_mut(&channel_id) {
+							updates.retain(|upd| upd.update_id != update.update_id);
+						}
+					}
 					self.apply_post_close_monitor_update(counterparty_node_id, channel_id, funding_txo, update);
 				},
 				BackgroundEvent::MonitorUpdatesComplete { counterparty_node_id, channel_id } => {
+					// We previously assumed background events would eventually be processed prior
+					// to another `ChannelManager` write, so we would immediately remove all
+					// in-flight monitor updates that completed since the last `ChannelManager`
+					// serialization. This isn't always the case, so we now keep them all around
+					// until we're ready to handle them, i.e., when `process_background_events` is
+					// called.
+					//
+					// Now that we can finally handle the background event, remove all in-flight
+					// monitor updates for this channel, as they have already been persisted to the
+					// monitor and can be applied to our internal state such that the channel
+					// resumes operation.
+					{
+						let per_peer_state = self.per_peer_state.read().unwrap();
+						let mut peer_state_lock;
+						let peer_state_mutex_opt = per_peer_state.get(&counterparty_node_id);
+						if peer_state_mutex_opt.is_none() { continue; }
+						peer_state_lock = peer_state_mutex_opt.unwrap().lock().unwrap();
+						let peer_state = &mut *peer_state_lock;
+						peer_state.in_flight_monitor_updates.remove(&channel_id);
+					}
 					self.channel_monitor_updated(&channel_id, None, &counterparty_node_id);
 				},
 			}
@@ -18134,39 +18175,47 @@ impl<
 			($counterparty_node_id: expr, $chan_in_flight_upds: expr, $monitor: expr,
 			 $peer_state: expr, $logger: expr, $channel_info_log: expr
 			) => { {
+				// We previously assumed background events would eventually be processed prior to
+				// another `ChannelManager` write, so we would immediately remove all in-flight
+				// monitor updates that completed since the last `ChannelManager` serialization.
+				// This isn't always the case, so we now keep them all around until we're ready to
+				// handle them, i.e., when `process_background_events` is called.
 				let mut max_in_flight_update_id = 0;
-				let starting_len =  $chan_in_flight_upds.len();
-				$chan_in_flight_upds.retain(|upd| upd.update_id > $monitor.get_latest_update_id());
-				if $chan_in_flight_upds.len() < starting_len {
+				let num_updates_completed = $chan_in_flight_upds
+					.iter()
+					.filter(|upd| upd.update_id <= $monitor.get_latest_update_id())
+					.count();
+				if num_updates_completed > 0 {
 					log_debug!(
 						$logger,
 						"{} ChannelMonitorUpdates completed after ChannelManager was last serialized",
-						starting_len - $chan_in_flight_upds.len()
+						num_updates_completed,
 					);
 				}
 				let funding_txo = $monitor.get_funding_txo();
-				for update in $chan_in_flight_upds.iter() {
-					log_debug!($logger, "Replaying ChannelMonitorUpdate {} for {}channel {}",
-						update.update_id, $channel_info_log, &$monitor.channel_id());
-					max_in_flight_update_id = cmp::max(max_in_flight_update_id, update.update_id);
-					pending_background_events.push(
-						BackgroundEvent::MonitorUpdateRegeneratedOnStartup {
-							counterparty_node_id: $counterparty_node_id,
-							funding_txo: funding_txo,
-							channel_id: $monitor.channel_id(),
-							update: update.clone(),
-						});
-				}
-				if $chan_in_flight_upds.is_empty() {
+				if num_updates_completed == $chan_in_flight_upds.len() {
 					// We had some updates to apply, but it turns out they had completed before we
 					// were serialized, we just weren't notified of that. Thus, we may have to run
 					// the completion actions for any monitor updates, but otherwise are done.
+					log_debug!($logger, "All monitor updates completed since the ChannelManager was last serialized");
 					pending_background_events.push(
 						BackgroundEvent::MonitorUpdatesComplete {
 							counterparty_node_id: $counterparty_node_id,
 							channel_id: $monitor.channel_id(),
 						});
 				} else {
+					for update in $chan_in_flight_upds.iter() {
+						log_debug!($logger, "Replaying ChannelMonitorUpdate {} for {}channel {}",
+							update.update_id, $channel_info_log, &$monitor.channel_id());
+						max_in_flight_update_id = cmp::max(max_in_flight_update_id, update.update_id);
+						pending_background_events.push(
+							BackgroundEvent::MonitorUpdateRegeneratedOnStartup {
+								counterparty_node_id: $counterparty_node_id,
+								funding_txo: funding_txo,
+								channel_id: $monitor.channel_id(),
+								update: update.clone(),
+							});
+					}
 					$peer_state.closed_channel_monitor_update_ids.entry($monitor.channel_id())
 						.and_modify(|v| *v = cmp::max(max_in_flight_update_id, *v))
 						.or_insert(max_in_flight_update_id);
@@ -18194,7 +18243,7 @@ impl<
 						.expect("We already checked for monitor presence when loading channels");
 					let mut max_in_flight_update_id = monitor.get_latest_update_id();
 					if let Some(in_flight_upds) = &mut in_flight_monitor_updates {
-						if let Some(mut chan_in_flight_upds) =
+						if let Some(chan_in_flight_upds) =
 							in_flight_upds.remove(&(*counterparty_id, *chan_id))
 						{
 							max_in_flight_update_id = cmp::max(
@@ -18238,7 +18287,7 @@ impl<
 		}
 
 		if let Some(in_flight_upds) = in_flight_monitor_updates {
-			for ((counterparty_id, channel_id), mut chan_in_flight_updates) in in_flight_upds {
+			for ((counterparty_id, channel_id), chan_in_flight_updates) in in_flight_upds {
 				let logger =
 					WithContext::from(&args.logger, Some(counterparty_id), Some(channel_id), None);
 				if let Some(monitor) = args.channel_monitors.get(&channel_id) {

lightning/src/ln/reload_tests.rs

@@ -11,7 +11,7 @@
 
 //! Functional tests which test for correct behavior across node restarts.
 
-use crate::chain::{ChannelMonitorUpdateStatus, Watch};
+use crate::chain::{ChannelMonitorUpdateStatus, Confirm, Watch};
 use crate::chain::chaininterface::LowerBoundedFeeEstimator;
 use crate::chain::channelmonitor::{ChannelMonitor, ChannelMonitorUpdateStep};
 use crate::routing::router::{PaymentParameters, RouteParameters};
@@ -37,6 +37,13 @@ use crate::prelude::*;
 
 use crate::ln::functional_test_utils::*;
 
+fn get_latest_mon_update_id<'a, 'b, 'c>(
+	node: &Node<'a, 'b, 'c>, channel_id: ChannelId,
+) -> (u64, u64) {
+	let monitor_id_state = node.chain_monitor.latest_monitor_update_id.lock().unwrap();
+	monitor_id_state.get(&channel_id).unwrap().clone()
+}
+
 #[test]
 fn test_funding_peer_disconnect() {
 	// Test that we can lock in our funding tx while disconnected
@@ -1566,3 +1573,84 @@ fn test_peer_storage() {
 	assert!(res.is_err());
 }
 
+#[test]
+fn test_hold_completed_inflight_monitor_updates_upon_manager_reload() {
+	// Test that if a `ChannelMonitorUpdate` completes after the `ChannelManager` is serialized,
+	// but before it is deserialized, we hold any completed in-flight updates until background event
+	// processing. Previously, we would remove completed monitor updates from
+	// `in_flight_monitor_updates` during deserialization, relying on
+	// [`ChannelManager::process_background_events`] to eventually be called before the
+	// `ChannelManager` is serialized again such that the channel is resumed and further updates can
+	// be made.
+	let chanmon_cfgs = create_chanmon_cfgs(2);
+	let node_cfgs = create_node_cfgs(2, &chanmon_cfgs);
+	let (persister_a, persister_b);
+	let (chain_monitor_a, chain_monitor_b);
+
+	let node_chanmgrs = create_node_chanmgrs(2, &node_cfgs, &[None, None]);
+	let nodes_0_deserialized_a;
+	let nodes_0_deserialized_b;
+
+	let mut nodes = create_network(2, &node_cfgs, &node_chanmgrs);
+	let chan_id = create_announced_chan_between_nodes(&nodes, 0, 1).2;
+
+	send_payment(&nodes[0], &[&nodes[1]], 1_000_000);
+
+	chanmon_cfgs[0].persister.set_update_ret(ChannelMonitorUpdateStatus::InProgress);
+
+	// Send a payment that will be pending due to an async monitor update.
+	let (route, payment_hash, _, payment_secret) =
+		get_route_and_payment_hash!(nodes[0], nodes[1], 1_000_000);
+	let payment_id = PaymentId(payment_hash.0);
+	let onion = RecipientOnionFields::secret_only(payment_secret);
+	nodes[0].node.send_payment_with_route(route, payment_hash, onion, payment_id).unwrap();
+	check_added_monitors(&nodes[0], 1);
+
+	assert!(nodes[0].node.get_and_clear_pending_msg_events().is_empty());
+	assert!(nodes[1].node.get_and_clear_pending_msg_events().is_empty());
+
+	// Serialize the ChannelManager while the monitor update is still in-flight.
+	let node_0_serialized = nodes[0].node.encode();
+
+	// Now complete the monitor update by calling force_channel_monitor_updated.
+	// This updates the monitor's state, but the ChannelManager still thinks it's pending.
+	let (_, latest_update_id) = get_latest_mon_update_id(&nodes[0], chan_id);
+	nodes[0].chain_monitor.chain_monitor.force_channel_monitor_updated(chan_id, latest_update_id);
+	let monitor_serialized_updated = get_monitor!(nodes[0], chan_id).encode();
+
+	// Reload the node with the updated monitor. Upon deserialization, the ChannelManager will
+	// detect that the monitor update completed (monitor's update_id >= the in-flight update_id)
+	// and queue a `BackgroundEvent::MonitorUpdatesComplete`.
+	nodes[0].node.peer_disconnected(nodes[1].node.get_our_node_id());
+	nodes[1].node.peer_disconnected(nodes[0].node.get_our_node_id());
+	reload_node!(
+		nodes[0],
+		test_default_channel_config(),
+		&node_0_serialized,
+		&[&monitor_serialized_updated[..]],
+		persister_a,
+		chain_monitor_a,
+		nodes_0_deserialized_a
+	);
+
+	// If we serialize again, even though we haven't processed any background events yet, we should
+	// still see the `BackgroundEvent::MonitorUpdatesComplete` be regenerated on startup.
+	let node_0_serialized = nodes[0].node.encode();
+	reload_node!(
+		nodes[0],
+		test_default_channel_config(),
+		&node_0_serialized,
+		&[&monitor_serialized_updated[..]],
+		persister_b,
+		chain_monitor_b,
+		nodes_0_deserialized_b
+	);
+
+	// Reconnect the nodes. We should finally see the `update_add_htlc` go out, as the reconnection
+	// should first process `BackgroundEvent::MonitorUpdatesComplete, allowing the channel to be
+	// resumed.
+	let mut reconnect_args = ReconnectArgs::new(&nodes[0], &nodes[1]);
+	reconnect_args.pending_htlc_adds = (0, 1);
+	reconnect_nodes(reconnect_args);
+}
+