+ }
+ let pending_outbounds = OutboundPayments {
+ pending_outbound_payments: Mutex::new(pending_outbound_payments.unwrap()),
+ retry_lock: Mutex::new(())
+ };
+
+ // We have to replay (or skip, if they were completed after we wrote the `ChannelManager`)
+ // each `ChannelMonitorUpdate` in `in_flight_monitor_updates`. After doing so, we have to
+ // check that each channel we have isn't newer than the latest `ChannelMonitorUpdate`(s) we
+ // replayed, and for each monitor update we have to replay we have to ensure there's a
+ // `ChannelMonitor` for it.
+ //
+ // In order to do so we first walk all of our live channels (so that we can check their
+ // state immediately after doing the update replays, when we have the `update_id`s
+ // available) and then walk any remaining in-flight updates.
+ //
+ // Because the actual handling of the in-flight updates is the same, it's macro'ized here:
+ let mut pending_background_events = Vec::new();
+ macro_rules! handle_in_flight_updates {
+ ($counterparty_node_id: expr, $chan_in_flight_upds: expr, $funding_txo: expr,
+ $monitor: expr, $peer_state: expr, $channel_info_log: expr
+ ) => { {
+ let mut max_in_flight_update_id = 0;
+ $chan_in_flight_upds.retain(|upd| upd.update_id > $monitor.get_latest_update_id());
+ for update in $chan_in_flight_upds.iter() {
+ log_trace!(args.logger, "Replaying ChannelMonitorUpdate {} for {}channel {}",
+ update.update_id, $channel_info_log, log_bytes!($funding_txo.to_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,
+ 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.
+ pending_background_events.push(
+ BackgroundEvent::MonitorUpdatesComplete {
+ counterparty_node_id: $counterparty_node_id,
+ channel_id: $funding_txo.to_channel_id(),
+ });
+ }
+ if $peer_state.in_flight_monitor_updates.insert($funding_txo, $chan_in_flight_upds).is_some() {
+ log_error!(args.logger, "Duplicate in-flight monitor update set for the same channel!");
+ return Err(DecodeError::InvalidValue);
+ }
+ max_in_flight_update_id
+ } }
+ }
+
+ for (counterparty_id, peer_state_mtx) in per_peer_state.iter_mut() {
+ let mut peer_state_lock = peer_state_mtx.lock().unwrap();
+ let peer_state = &mut *peer_state_lock;
+ for (_, chan) in peer_state.channel_by_id.iter() {
+ // Channels that were persisted have to be funded, otherwise they should have been
+ // discarded.
+ let funding_txo = chan.context.get_funding_txo().ok_or(DecodeError::InvalidValue)?;
+ let monitor = args.channel_monitors.get(&funding_txo)
+ .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) = in_flight_upds.remove(&(*counterparty_id, funding_txo)) {
+ max_in_flight_update_id = cmp::max(max_in_flight_update_id,
+ handle_in_flight_updates!(*counterparty_id, chan_in_flight_upds,
+ funding_txo, monitor, peer_state, ""));
+ }
+ }
+ if chan.get_latest_unblocked_monitor_update_id() > max_in_flight_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 {} with update_id through {} in-flight",
+ log_bytes!(chan.context.channel_id()), monitor.get_latest_update_id(), max_in_flight_update_id);
+ log_error!(args.logger, " but the ChannelManager is at update_id {}.", chan.get_latest_unblocked_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.");
+ log_error!(args.logger, " Please ensure the chain::Watch API requirements are met and file a bug report at https://github.com/lightningdevkit/rust-lightning");
+ return Err(DecodeError::InvalidValue);
+ }
+ }
+ }
+
+ if let Some(in_flight_upds) = in_flight_monitor_updates {
+ for ((counterparty_id, funding_txo), mut chan_in_flight_updates) in in_flight_upds {
+ if let Some(monitor) = args.channel_monitors.get(&funding_txo) {
+ // Now that we've removed all the in-flight monitor updates for channels that are
+ // still open, we need to replay any monitor updates that are for closed channels,
+ // creating the neccessary peer_state entries as we go.
+ let peer_state_mutex = per_peer_state.entry(counterparty_id).or_insert_with(|| {
+ Mutex::new(peer_state_from_chans(HashMap::new()))
+ });
+ let mut peer_state = peer_state_mutex.lock().unwrap();
+ handle_in_flight_updates!(counterparty_id, chan_in_flight_updates,
+ funding_txo, monitor, peer_state, "closed ");
+ } else {
+ log_error!(args.logger, "A ChannelMonitor is missing even though we have in-flight updates for it! This indicates a potentially-critical violation of the chain::Watch API!");
+ log_error!(args.logger, " The ChannelMonitor for channel {} is missing.",
+ log_bytes!(funding_txo.to_channel_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.");
+ log_error!(args.logger, " Please ensure the chain::Watch API requirements are met and file a bug report at https://github.com/lightningdevkit/rust-lightning");
+ return Err(DecodeError::InvalidValue);
+ }
+ }
+ }
+
+ // Note that we have to do the above replays before we push new monitor updates.
+ pending_background_events.append(&mut close_background_events);
+
+ // If there's any preimages for forwarded HTLCs hanging around in ChannelMonitors we
+ // should ensure we try them again on the inbound edge. We put them here and do so after we
+ // have a fully-constructed `ChannelManager` at the end.
+ let mut pending_claims_to_replay = Vec::new();
+
+ {