Hold in-flight monitor updates until background event processing

diff --git a/lightning/src/ln/channelmanager.rs b/lightning/src/ln/channelmanager.rs
index 532514a..3656a03 100644
--- a/lightning/src/ln/channelmanager.rs
+++ b/lightning/src/ln/channelmanager.rs
@@ -1277,7 +1277,11 @@ enum BackgroundEvent {
 	/// Some [`ChannelMonitorUpdate`] (s) completed before we were serialized but we still have
 	/// them marked pending, thus we need to run any [`MonitorUpdateCompletionAction`] (s) pending
 	/// on a channel.
-	MonitorUpdatesComplete { counterparty_node_id: PublicKey, channel_id: ChannelId },
+	MonitorUpdatesComplete {
+		counterparty_node_id: PublicKey,
+		channel_id: ChannelId,
+		highest_update_id_completed: u64,
+	},
 }
 
 /// A pointer to a channel that is unblocked when an event is surfaced
@@ -8153,8 +8157,21 @@ impl<
 				BackgroundEvent::MonitorUpdateRegeneratedOnStartup { counterparty_node_id, funding_txo, channel_id, update } => {
 					self.apply_post_close_monitor_update(counterparty_node_id, channel_id, funding_txo, update);
 				},
-				BackgroundEvent::MonitorUpdatesComplete { counterparty_node_id, channel_id } => {
-					self.channel_monitor_updated(&channel_id, None, &counterparty_node_id);
+				BackgroundEvent::MonitorUpdatesComplete {
+					counterparty_node_id,
+					channel_id,
+					highest_update_id_completed,
+				} => {
+					// Now that we can finally handle the background event, remove all in-flight
+					// monitor updates for this channel that we've known to complete, as they have
+					// already been persisted to the monitor and can be applied to our internal
+					// state such that the channel resumes operation if no new updates have been
+					// made since.
+					self.channel_monitor_updated(
+						&channel_id,
+						Some(highest_update_id_completed),
+						&counterparty_node_id,
+					);
 				},
 			}
 		}
@@ -18134,39 +18151,58 @@ impl<
 			($counterparty_node_id: expr, $chan_in_flight_upds: expr, $monitor: expr,
 			 $peer_state: expr, $logger: expr, $channel_info_log: expr
 			) => { {
+				// When all in-flight updates have completed after we were last serialized, we
+				// need to remove them. However, we can't guarantee that the next serialization
+				// will have happened after processing the
+				// `BackgroundEvent::MonitorUpdatesComplete`, so removing them now could lead to the
+				// channel never being resumed as the event would not be regenerated after another
+				// reload. At the same time, we don't want to resume the channel now because there
+				// may be post-update actions to handle. Therefore, we're forced to keep tracking
+				// the completed in-flight updates (but only when they have all completed) until we
+				// are processing the `BackgroundEvent::MonitorUpdatesComplete`.
 				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(|update| {
+						max_in_flight_update_id = cmp::max(max_in_flight_update_id, update.update_id);
+						update.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 all_updates_completed = num_updates_completed == $chan_in_flight_upds.len();
+
 				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() {
-					// 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.
+				if all_updates_completed {
+					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(),
+							highest_update_id_completed: max_in_flight_update_id,
 						});
 				} else {
+					$chan_in_flight_upds.retain(|update| {
+						let replay = update.update_id > $monitor.get_latest_update_id();
+						if replay {
+							log_debug!($logger, "Replaying ChannelMonitorUpdate {} for {}channel {}",
+								update.update_id, $channel_info_log, &$monitor.channel_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(),
+								}
+							);
+						}
+						replay
+					});
 					$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);
diff --git a/lightning/src/ln/reload_tests.rs b/lightning/src/ln/reload_tests.rs
index a8206df..c043205 100644
--- a/lightning/src/ln/reload_tests.rs
+++ b/lightning/src/ln/reload_tests.rs
@@ -1566,3 +1566,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) = nodes[0].chain_monitor.get_latest_mon_update_id(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);
+}
+