Reapply pending ChannelMonitorUpdates on startup

If a `ChannelMonitorUpdate` was created and given to the user but
left uncompleted when the `ChannelManager` is persisted prior to a
restart, the user likely lost the `ChannelMonitorUpdate`(s). Thus,
we need to replay them for the user, which we do here using the
new `BackgroundEvent::MonitorUpdateRegeneratedOnStartup` variant.

Author: TheBlueMatt

lightning/src/ln/channel.rs

@@ -5050,10 +5050,25 @@ impl<Signer: WriteableEcdsaChannelSigner> Channel<Signer> {
 		self.pending_monitor_updates.is_empty()
 	}
 
+	pub fn complete_all_mon_updates_through(&mut self, update_id: u64) {
+		self.pending_monitor_updates.retain(|upd| {
+			if upd.update.update_id <= update_id {
+				assert!(!upd.blocked, "Completed update must have flown");
+				false
+			} else { true }
+		});
+	}
+
 	pub fn complete_one_mon_update(&mut self, update_id: u64) {
 		self.pending_monitor_updates.retain(|upd| upd.update.update_id != update_id);
 	}
 
+	/// Returns an iterator over all unblocked monitor updates which have not yet completed.
+	pub fn uncompleted_unblocked_mon_updates(&self) -> impl Iterator<Item=&ChannelMonitorUpdate> {
+		self.pending_monitor_updates.iter()
+			.filter_map(|upd| if upd.blocked { None } else { Some(&upd.update) })
+	}
+
 	/// Returns true if funding_created was sent/received.
 	pub fn is_funding_initiated(&self) -> bool {
 		self.channel_state >= ChannelState::FundingSent as u32

lightning/src/ln/channelmanager.rs

@@ -7882,7 +7882,10 @@ where
 						}
 					}
 				} else {
-					log_info!(args.logger, "Successfully loaded channel {}", log_bytes!(channel.channel_id()));
+					log_info!(args.logger, "Successfully loaded channel {} at update_id {} against monitor at update id {}",
+						log_bytes!(channel.channel_id()), channel.get_latest_monitor_update_id(),
+						monitor.get_latest_update_id());
+					channel.complete_all_mon_updates_through(monitor.get_latest_update_id());
 					if let Some(short_channel_id) = channel.get_short_channel_id() {
 						short_to_chan_info.insert(short_channel_id, (channel.get_counterparty_node_id(), channel.channel_id()));
 					}
@@ -7996,6 +7999,24 @@ where
 			}
 		}
 
+		for (node_id, peer_mtx) in per_peer_state.iter() {
+			let peer_state = peer_mtx.lock().unwrap();
+			for (_, chan) in peer_state.channel_by_id.iter() {
+				for update in chan.uncompleted_unblocked_mon_updates() {
+					if let Some(funding_txo) = chan.get_funding_txo() {
+						log_trace!(args.logger, "Replaying ChannelMonitorUpdate {} for channel {}",
+							update.update_id, log_bytes!(funding_txo.to_channel_id()));
+						pending_background_events.push(
+							BackgroundEvent::MonitorUpdateRegeneratedOnStartup {
+								counterparty_node_id: *node_id, funding_txo, update: update.clone(),
+							});
+					} else {
+						return Err(DecodeError::InvalidValue);
+					}
+				}
+			}
+		}
+
 		let _last_node_announcement_serial: u32 = Readable::read(reader)?; // Only used < 0.0.111
 		let highest_seen_timestamp: u32 = Readable::read(reader)?;