Store + process pending ChannelMonitorUpdates in Channel

The previous commits set up the ability for us to hold
`ChannelMonitorUpdate`s which are pending until we're ready to pass
them to users and have them be applied. However, if the
`ChannelManager` is persisted while we're waiting to give the user
a `ChannelMonitorUpdate` we'll be confused on restart - seeing our
latest `ChannelMonitor` state as stale compared to our
`ChannelManager` - a critical error.

Luckily the solution is trivial, we simply need to store the
pending `ChannelMonitorUpdate` state and load it with the
`ChannelManager` data, allowing stale monitors on load as long as
we have the missing pending updates between where we are and the
latest `ChannelMonitor` state.

Author: TheBlueMatt

lightning/src/ln/channel.rs

@@ -489,6 +489,11 @@ struct PendingChannelMonitorUpdate {
 	blocked: bool,
 }
 
+impl_writeable_tlv_based!(PendingChannelMonitorUpdate, {
+	(0, update, required),
+	(2, flown, required),
+});
+
 // TODO: We should refactor this to be an Inbound/OutboundChannel until initial setup handshaking
 // has been completed, and then turn into a Channel to get compiler-time enforcement of things like
 // calling channel_id() before we're set up or things like get_outbound_funding_signed on an
@@ -4993,6 +4998,11 @@ impl<Signer: WriteableEcdsaChannelSigner> Channel<Signer> {
 		(self.channel_state & ChannelState::MonitorUpdateInProgress as u32) != 0
 	}
 
+	pub fn get_latest_complete_monitor_update_id(&self) -> u64 {
+		if self.pending_monitor_updates.is_empty() { return self.get_latest_monitor_update_id(); }
+		self.pending_monitor_updates[0].update.update_id - 1
+	}
+
 	/// Returns the next blocked monitor update, if one exists, and a bool which indicates a
 	/// further blocked monitor update exists after the next.
 	pub fn unblock_next_blocked_monitor_update(&mut self) -> Option<(&ChannelMonitorUpdate, bool)> {
@@ -6591,6 +6601,7 @@ impl<Signer: WriteableEcdsaChannelSigner> Writeable for Channel<Signer> {
 			(27, self.channel_keys_id, required),
 			(29, self.temporary_channel_id, option),
 			(31, channel_pending_event_emitted, option),
+			(33, self.pending_monitor_updates, vec_type),
 		});
 
 		Ok(())
@@ -6865,6 +6876,8 @@ impl<'a, 'b, 'c, ES: Deref, SP: Deref> ReadableArgs<(&'a ES, &'b SP, u32, &'c Ch
 		let mut channel_keys_id: Option<[u8; 32]> = None;
 		let mut temporary_channel_id: Option<[u8; 32]> = None;
 
+		let mut pending_monitor_updates = Some(Vec::new());
+
 		read_tlv_fields!(reader, {
 			(0, announcement_sigs, option),
 			(1, minimum_depth, option),
@@ -6886,6 +6899,7 @@ impl<'a, 'b, 'c, ES: Deref, SP: Deref> ReadableArgs<(&'a ES, &'b SP, u32, &'c Ch
 			(27, channel_keys_id, option),
 			(29, temporary_channel_id, option),
 			(31, channel_pending_event_emitted, option),
+			(33, pending_monitor_updates, vec_type),
 		});
 
 		let (channel_keys_id, holder_signer) = if let Some(channel_keys_id) = channel_keys_id {
@@ -7052,7 +7066,7 @@ impl<'a, 'b, 'c, ES: Deref, SP: Deref> ReadableArgs<(&'a ES, &'b SP, u32, &'c Ch
 			channel_type: channel_type.unwrap(),
 			channel_keys_id,
 
-			pending_monitor_updates: Vec::new(),
+			pending_monitor_updates: pending_monitor_updates.unwrap(),
 		})
 	}
 }

lightning/src/ln/channelmanager.rs

@@ -7560,14 +7560,11 @@ where
 			let funding_txo = channel.get_funding_txo().ok_or(DecodeError::InvalidValue)?;
 			funding_txo_set.insert(funding_txo.clone());
 			if let Some(ref mut monitor) = args.channel_monitors.get_mut(&funding_txo) {
-				if channel.get_cur_holder_commitment_transaction_number() < monitor.get_cur_holder_commitment_number() ||
-						channel.get_revoked_counterparty_commitment_transaction_number() < monitor.get_min_seen_secret() ||
-						channel.get_cur_counterparty_commitment_transaction_number() < monitor.get_cur_counterparty_commitment_number() ||
-						channel.get_latest_monitor_update_id() > monitor.get_latest_update_id() {
+				if channel.get_latest_complete_monitor_update_id() > monitor.get_latest_update_id() {
 					// If the channel is ahead of the monitor, return InvalidValue:
 					log_error!(args.logger, "A ChannelMonitor is stale compared to the current ChannelManager! This indicates a potentially-critical violation of the chain::Watch API!");
 					log_error!(args.logger, " The ChannelMonitor for channel {} is at update_id {} but the ChannelManager is at update_id {}.",
-						log_bytes!(channel.channel_id()), monitor.get_latest_update_id(), channel.get_latest_monitor_update_id());
+						log_bytes!(channel.channel_id()), monitor.get_latest_update_id(), channel.get_latest_complete_monitor_update_id());
 					log_error!(args.logger, " The chain::Watch API *requires* that monitors are persisted durably before returning,");
 					log_error!(args.logger, " client applications must ensure that ChannelMonitor data is always available and the latest to avoid funds loss!");
 					log_error!(args.logger, " Without the latest ChannelMonitor we cannot continue without risking funds.");