Log shutdown including which side of the channel initiated shutdown
[rust-lightning] / lightning / src / ln / channelmanager.rs
index d8d79b9acd2d48238df33ce642c501c3997cd99b..a50c072b4cbd07b8740a3c9a69af54c01f44f3b1 100644 (file)
@@ -37,14 +37,14 @@ use bitcoin::secp256k1;
 
 use chain;
 use chain::{Confirm, Watch, BestBlock};
-use chain::chaininterface::{BroadcasterInterface, FeeEstimator};
+use chain::chaininterface::{BroadcasterInterface, ConfirmationTarget, FeeEstimator};
 use chain::channelmonitor::{ChannelMonitor, ChannelMonitorUpdate, ChannelMonitorUpdateStep, ChannelMonitorUpdateErr, HTLC_FAIL_BACK_BUFFER, CLTV_CLAIM_BUFFER, LATENCY_GRACE_PERIOD_BLOCKS, ANTI_REORG_DELAY, MonitorEvent, CLOSED_CHANNEL_UPDATE_ID};
 use chain::transaction::{OutPoint, TransactionData};
 // Since this struct is returned in `list_channels` methods, expose it here in case users want to
 // construct one themselves.
 use ln::{PaymentHash, PaymentPreimage, PaymentSecret};
 pub use ln::channel::CounterpartyForwardingInfo;
-use ln::channel::{Channel, ChannelError, ChannelUpdateStatus};
+use ln::channel::{Channel, ChannelError, ChannelUpdateStatus, UpdateFulfillCommitFetch};
 use ln::features::{InitFeatures, NodeFeatures};
 use routing::router::{Route, RouteHop};
 use ln::msgs;
@@ -57,20 +57,20 @@ use util::events::{EventHandler, EventsProvider, MessageSendEvent, MessageSendEv
 use util::{byte_utils, events};
 use util::ser::{Readable, ReadableArgs, MaybeReadable, Writeable, Writer};
 use util::chacha20::{ChaCha20, ChaChaReader};
-use util::logger::Logger;
+use util::logger::{Logger, Level};
 use util::errors::APIError;
 
+use io;
 use prelude::*;
 use core::{cmp, mem};
 use core::cell::RefCell;
-use std::io::{Cursor, Read};
+use io::{Cursor, Read};
 use sync::{Arc, Condvar, Mutex, MutexGuard, RwLock, RwLockReadGuard};
 use core::sync::atomic::{AtomicUsize, Ordering};
 use core::time::Duration;
 #[cfg(any(test, feature = "allow_wallclock_use"))]
 use std::time::Instant;
 use core::ops::Deref;
-use bitcoin::hashes::hex::ToHex;
 
 // We hold various information about HTLC relay in the HTLC objects in Channel itself:
 //
@@ -207,6 +207,14 @@ pub(super) enum HTLCFailReason {
        }
 }
 
+/// Return value for claim_funds_from_hop
+enum ClaimFundsFromHop {
+       PrevHopForceClosed,
+       MonitorUpdateFail(PublicKey, MsgHandleErrInternal, Option<u64>),
+       Success(u64),
+       DuplicateClaim,
+}
+
 type ShutdownResult = (Option<(OutPoint, ChannelMonitorUpdate)>, Vec<(HTLCSource, PaymentHash)>);
 
 /// Error type returned across the channel_state mutex boundary. When an Err is generated for a
@@ -268,6 +276,10 @@ impl MsgHandleErrInternal {
        fn from_chan_no_close(err: ChannelError, channel_id: [u8; 32]) -> Self {
                Self {
                        err: match err {
+                               ChannelError::Warn(msg) =>  LightningError {
+                                       err: msg,
+                                       action: msgs::ErrorAction::IgnoreError,
+                               },
                                ChannelError::Ignore(msg) => LightningError {
                                        err: msg,
                                        action: msgs::ErrorAction::IgnoreError,
@@ -482,6 +494,8 @@ pub struct ChannelManager<Signer: Sign, M: Deref, T: Deref, K: Deref, F: Deref,
        /// Because adding or removing an entry is rare, we usually take an outer read lock and then
        /// operate on the inner value freely. Sadly, this prevents parallel operation when opening a
        /// new channel.
+       ///
+       /// If also holding `channel_state` lock, must lock `channel_state` prior to `per_peer_state`.
        per_peer_state: RwLock<HashMap<PublicKey, Mutex<PeerState>>>,
 
        pending_events: Mutex<Vec<events::Event>>,
@@ -809,6 +823,11 @@ macro_rules! handle_error {
 macro_rules! convert_chan_err {
        ($self: ident, $err: expr, $short_to_id: expr, $channel: expr, $channel_id: expr) => {
                match $err {
+                       ChannelError::Warn(msg) => {
+                               //TODO: Once warning messages are merged, we should send a `warning` message to our
+                               //peer here.
+                               (false, MsgHandleErrInternal::from_chan_no_close(ChannelError::Ignore(msg), $channel_id.clone()))
+                       },
                        ChannelError::Ignore(msg) => {
                                (false, MsgHandleErrInternal::from_chan_no_close(ChannelError::Ignore(msg), $channel_id.clone()))
                        },
@@ -862,6 +881,18 @@ macro_rules! try_chan_entry {
        }
 }
 
+macro_rules! remove_channel {
+       ($channel_state: expr, $entry: expr) => {
+               {
+                       let channel = $entry.remove_entry().1;
+                       if let Some(short_id) = channel.get_short_channel_id() {
+                               $channel_state.short_to_id.remove(&short_id);
+                       }
+                       channel
+               }
+       }
+}
+
 macro_rules! handle_monitor_err {
        ($self: ident, $err: expr, $channel_state: expr, $entry: expr, $action_type: path, $resend_raa: expr, $resend_commitment: expr) => {
                handle_monitor_err!($self, $err, $channel_state, $entry, $action_type, $resend_raa, $resend_commitment, Vec::new(), Vec::new())
@@ -1156,8 +1187,18 @@ impl<Signer: Sign, M: Deref, T: Deref, K: Deref, F: Deref, L: Deref> ChannelMana
                        return Err(APIError::APIMisuseError { err: format!("Channel value must be at least 1000 satoshis. It was {}", channel_value_satoshis) });
                }
 
-               let config = if override_config.is_some() { override_config.as_ref().unwrap() } else { &self.default_configuration };
-               let channel = Channel::new_outbound(&self.fee_estimator, &self.keys_manager, their_network_key, channel_value_satoshis, push_msat, user_id, config)?;
+               let channel = {
+                       let per_peer_state = self.per_peer_state.read().unwrap();
+                       match per_peer_state.get(&their_network_key) {
+                               Some(peer_state) => {
+                                       let peer_state = peer_state.lock().unwrap();
+                                       let their_features = &peer_state.latest_features;
+                                       let config = if override_config.is_some() { override_config.as_ref().unwrap() } else { &self.default_configuration };
+                                       Channel::new_outbound(&self.fee_estimator, &self.keys_manager, their_network_key, their_features, channel_value_satoshis, push_msat, user_id, config)?
+                               },
+                               None => return Err(APIError::ChannelUnavailable { err: format!("Not connected to node: {}", their_network_key) }),
+                       }
+               };
                let res = channel.get_open_channel(self.genesis_hash.clone());
 
                let _persistence_guard = PersistenceNotifierGuard::notify_on_drop(&self.total_consistency_lock, &self.persistence_notifier);
@@ -1251,40 +1292,61 @@ impl<Signer: Sign, M: Deref, T: Deref, K: Deref, F: Deref, L: Deref> ChannelMana
        pub fn close_channel(&self, channel_id: &[u8; 32]) -> Result<(), APIError> {
                let _persistence_guard = PersistenceNotifierGuard::notify_on_drop(&self.total_consistency_lock, &self.persistence_notifier);
 
-               let (mut failed_htlcs, chan_option) = {
+               let counterparty_node_id;
+               let mut failed_htlcs: Vec<(HTLCSource, PaymentHash)>;
+               let result: Result<(), _> = loop {
                        let mut channel_state_lock = self.channel_state.lock().unwrap();
                        let channel_state = &mut *channel_state_lock;
                        match channel_state.by_id.entry(channel_id.clone()) {
                                hash_map::Entry::Occupied(mut chan_entry) => {
-                                       let (shutdown_msg, failed_htlcs) = chan_entry.get_mut().get_shutdown()?;
+                                       counterparty_node_id = chan_entry.get().get_counterparty_node_id();
+                                       let per_peer_state = self.per_peer_state.read().unwrap();
+                                       let (shutdown_msg, monitor_update, htlcs) = match per_peer_state.get(&counterparty_node_id) {
+                                               Some(peer_state) => {
+                                                       let peer_state = peer_state.lock().unwrap();
+                                                       let their_features = &peer_state.latest_features;
+                                                       chan_entry.get_mut().get_shutdown(&self.keys_manager, their_features)?
+                                               },
+                                               None => return Err(APIError::ChannelUnavailable { err: format!("Not connected to node: {}", counterparty_node_id) }),
+                                       };
+                                       failed_htlcs = htlcs;
+
+                                       // Update the monitor with the shutdown script if necessary.
+                                       if let Some(monitor_update) = monitor_update {
+                                               if let Err(e) = self.chain_monitor.update_channel(chan_entry.get().get_funding_txo().unwrap(), monitor_update) {
+                                                       let (result, is_permanent) =
+                                                               handle_monitor_err!(self, e, channel_state.short_to_id, chan_entry.get_mut(), RAACommitmentOrder::CommitmentFirst, false, false, Vec::new(), Vec::new(), chan_entry.key());
+                                                       if is_permanent {
+                                                               remove_channel!(channel_state, chan_entry);
+                                                               break result;
+                                                       }
+                                               }
+                                       }
+
                                        channel_state.pending_msg_events.push(events::MessageSendEvent::SendShutdown {
-                                               node_id: chan_entry.get().get_counterparty_node_id(),
+                                               node_id: counterparty_node_id,
                                                msg: shutdown_msg
                                        });
+
                                        if chan_entry.get().is_shutdown() {
-                                               if let Some(short_id) = chan_entry.get().get_short_channel_id() {
-                                                       channel_state.short_to_id.remove(&short_id);
+                                               let channel = remove_channel!(channel_state, chan_entry);
+                                               if let Ok(channel_update) = self.get_channel_update_for_broadcast(&channel) {
+                                                       channel_state.pending_msg_events.push(events::MessageSendEvent::BroadcastChannelUpdate {
+                                                               msg: channel_update
+                                                       });
                                                }
-                                               (failed_htlcs, Some(chan_entry.remove_entry().1))
-                                       } else { (failed_htlcs, None) }
+                                       }
+                                       break Ok(());
                                },
                                hash_map::Entry::Vacant(_) => return Err(APIError::ChannelUnavailable{err: "No such channel".to_owned()})
                        }
                };
+
                for htlc_source in failed_htlcs.drain(..) {
                        self.fail_htlc_backwards_internal(self.channel_state.lock().unwrap(), htlc_source.0, &htlc_source.1, HTLCFailReason::Reason { failure_code: 0x4000 | 8, data: Vec::new() });
                }
-               let chan_update = if let Some(chan) = chan_option {
-                       self.get_channel_update_for_broadcast(&chan).ok()
-               } else { None };
-
-               if let Some(update) = chan_update {
-                       let mut channel_state = self.channel_state.lock().unwrap();
-                       channel_state.pending_msg_events.push(events::MessageSendEvent::BroadcastChannelUpdate {
-                               msg: update
-                       });
-               }
 
+               let _ = handle_error!(self, result, counterparty_node_id);
                Ok(())
        }
 
@@ -1465,10 +1527,11 @@ impl<Signer: Sign, M: Deref, T: Deref, K: Deref, F: Deref, L: Deref> ChannelMana
 
                        // OUR PAYMENT!
                        // final_expiry_too_soon
-                       // We have to have some headroom to broadcast on chain if we have the preimage, so make sure we have at least
-                       // HTLC_FAIL_BACK_BUFFER blocks to go.
-                       // Also, ensure that, in the case of an unknown payment hash, our payment logic has enough time to fail the HTLC backward
-                       // before our onchain logic triggers a channel closure (see HTLC_FAIL_BACK_BUFFER rational).
+                       // We have to have some headroom to broadcast on chain if we have the preimage, so make sure
+                       // we have at least HTLC_FAIL_BACK_BUFFER blocks to go.
+                       // Also, ensure that, in the case of an unknown preimage for the received payment hash, our
+                       // payment logic has enough time to fail the HTLC backward before our onchain logic triggers a
+                       // channel closure (see HTLC_FAIL_BACK_BUFFER rationale).
                        if (msg.cltv_expiry as u64) <= self.best_block.read().unwrap().height() as u64 + HTLC_FAIL_BACK_BUFFER as u64 + 1 {
                                return_err!("The final CLTV expiry is too soon to handle", 17, &[0;0]);
                        }
@@ -1848,6 +1911,9 @@ impl<Signer: Sign, M: Deref, T: Deref, K: Deref, F: Deref, L: Deref> ChannelMana
                        // for now more than 10 paths likely carries too much one-path failure.
                        return Err(PaymentSendFailure::ParameterError(APIError::RouteError{err: "Sending over more than 10 paths is not currently supported"}));
                }
+               if payment_secret.is_none() && route.paths.len() > 1 {
+                       return Err(PaymentSendFailure::ParameterError(APIError::APIMisuseError{err: "Payment secret is required for multi-path payments".to_string()}));
+               }
                let mut total_value = 0;
                let our_node_id = self.get_our_node_id();
                let mut path_errs = Vec::with_capacity(route.paths.len());
@@ -1902,9 +1968,13 @@ impl<Signer: Sign, M: Deref, T: Deref, K: Deref, F: Deref, L: Deref> ChannelMana
        /// would be able to guess -- otherwise, an intermediate node may claim the payment and it will
        /// never reach the recipient.
        ///
+       /// See [`send_payment`] documentation for more details on the return value of this function.
+       ///
        /// Similar to regular payments, you MUST NOT reuse a `payment_preimage` value. See
        /// [`send_payment`] for more information about the risks of duplicate preimage usage.
        ///
+       /// Note that `route` must have exactly one path.
+       ///
        /// [`send_payment`]: Self::send_payment
        pub fn send_spontaneous_payment(&self, route: &Route, payment_preimage: Option<PaymentPreimage>) -> Result<PaymentHash, PaymentSendFailure> {
                let preimage = match payment_preimage {
@@ -2261,9 +2331,9 @@ impl<Signer: Sign, M: Deref, T: Deref, K: Deref, F: Deref, L: Deref> ChannelMana
                                                                        // close channel and then send error message to peer.
                                                                        let counterparty_node_id = chan.get().get_counterparty_node_id();
                                                                        let err: Result<(), _>  = match e {
-                                                                               ChannelError::Ignore(_) => {
+                                                                               ChannelError::Ignore(_) | ChannelError::Warn(_) => {
                                                                                        panic!("Stated return value requirements in send_commitment() were not met");
-                                                                               },
+                                                                               }
                                                                                ChannelError::Close(msg) => {
                                                                                        log_trace!(self.logger, "Closing channel {} due to Close-required error: {}", log_bytes!(chan.key()[..]), msg);
                                                                                        let (channel_id, mut channel) = chan.remove_entry();
@@ -2499,46 +2569,151 @@ impl<Signer: Sign, M: Deref, T: Deref, K: Deref, F: Deref, L: Deref> ChannelMana
                self.process_background_events();
        }
 
-       /// If a peer is disconnected we mark any channels with that peer as 'disabled'.
-       /// After some time, if channels are still disabled we need to broadcast a ChannelUpdate
-       /// to inform the network about the uselessness of these channels.
+       fn update_channel_fee(&self, short_to_id: &mut HashMap<u64, [u8; 32]>, pending_msg_events: &mut Vec<events::MessageSendEvent>, chan_id: &[u8; 32], chan: &mut Channel<Signer>, new_feerate: u32) -> (bool, NotifyOption, Result<(), MsgHandleErrInternal>) {
+               if !chan.is_outbound() { return (true, NotifyOption::SkipPersist, Ok(())); }
+               // If the feerate has decreased by less than half, don't bother
+               if new_feerate <= chan.get_feerate() && new_feerate * 2 > chan.get_feerate() {
+                       log_trace!(self.logger, "Channel {} does not qualify for a feerate change from {} to {}.",
+                               log_bytes!(chan_id[..]), chan.get_feerate(), new_feerate);
+                       return (true, NotifyOption::SkipPersist, Ok(()));
+               }
+               if !chan.is_live() {
+                       log_trace!(self.logger, "Channel {} does not qualify for a feerate change from {} to {} as it cannot currently be updated (probably the peer is disconnected).",
+                               log_bytes!(chan_id[..]), chan.get_feerate(), new_feerate);
+                       return (true, NotifyOption::SkipPersist, Ok(()));
+               }
+               log_trace!(self.logger, "Channel {} qualifies for a feerate change from {} to {}.",
+                       log_bytes!(chan_id[..]), chan.get_feerate(), new_feerate);
+
+               let mut retain_channel = true;
+               let res = match chan.send_update_fee_and_commit(new_feerate, &self.logger) {
+                       Ok(res) => Ok(res),
+                       Err(e) => {
+                               let (drop, res) = convert_chan_err!(self, e, short_to_id, chan, chan_id);
+                               if drop { retain_channel = false; }
+                               Err(res)
+                       }
+               };
+               let ret_err = match res {
+                       Ok(Some((update_fee, commitment_signed, monitor_update))) => {
+                               if let Err(e) = self.chain_monitor.update_channel(chan.get_funding_txo().unwrap(), monitor_update) {
+                                       let (res, drop) = handle_monitor_err!(self, e, short_to_id, chan, RAACommitmentOrder::CommitmentFirst, false, true, Vec::new(), Vec::new(), chan_id);
+                                       if drop { retain_channel = false; }
+                                       res
+                               } else {
+                                       pending_msg_events.push(events::MessageSendEvent::UpdateHTLCs {
+                                               node_id: chan.get_counterparty_node_id(),
+                                               updates: msgs::CommitmentUpdate {
+                                                       update_add_htlcs: Vec::new(),
+                                                       update_fulfill_htlcs: Vec::new(),
+                                                       update_fail_htlcs: Vec::new(),
+                                                       update_fail_malformed_htlcs: Vec::new(),
+                                                       update_fee: Some(update_fee),
+                                                       commitment_signed,
+                                               },
+                                       });
+                                       Ok(())
+                               }
+                       },
+                       Ok(None) => Ok(()),
+                       Err(e) => Err(e),
+               };
+               (retain_channel, NotifyOption::DoPersist, ret_err)
+       }
+
+       #[cfg(fuzzing)]
+       /// In chanmon_consistency we want to sometimes do the channel fee updates done in
+       /// timer_tick_occurred, but we can't generate the disabled channel updates as it considers
+       /// these a fuzz failure (as they usually indicate a channel force-close, which is exactly what
+       /// it wants to detect). Thus, we have a variant exposed here for its benefit.
+       pub fn maybe_update_chan_fees(&self) {
+               PersistenceNotifierGuard::optionally_notify(&self.total_consistency_lock, &self.persistence_notifier, || {
+                       let mut should_persist = NotifyOption::SkipPersist;
+
+                       let new_feerate = self.fee_estimator.get_est_sat_per_1000_weight(ConfirmationTarget::Normal);
+
+                       let mut handle_errors = Vec::new();
+                       {
+                               let mut channel_state_lock = self.channel_state.lock().unwrap();
+                               let channel_state = &mut *channel_state_lock;
+                               let pending_msg_events = &mut channel_state.pending_msg_events;
+                               let short_to_id = &mut channel_state.short_to_id;
+                               channel_state.by_id.retain(|chan_id, chan| {
+                                       let (retain_channel, chan_needs_persist, err) = self.update_channel_fee(short_to_id, pending_msg_events, chan_id, chan, new_feerate);
+                                       if chan_needs_persist == NotifyOption::DoPersist { should_persist = NotifyOption::DoPersist; }
+                                       if err.is_err() {
+                                               handle_errors.push(err);
+                                       }
+                                       retain_channel
+                               });
+                       }
+
+                       should_persist
+               });
+       }
+
+       /// Performs actions which should happen on startup and roughly once per minute thereafter.
        ///
-       /// This method handles all the details, and must be called roughly once per minute.
+       /// This currently includes:
+       ///  * Increasing or decreasing the on-chain feerate estimates for our outbound channels,
+       ///  * Broadcasting `ChannelUpdate` messages if we've been disconnected from our peer for more
+       ///    than a minute, informing the network that they should no longer attempt to route over
+       ///    the channel.
        ///
-       /// Note that in some rare cases this may generate a `chain::Watch::update_channel` call.
+       /// Note that this may cause reentrancy through `chain::Watch::update_channel` calls or feerate
+       /// estimate fetches.
        pub fn timer_tick_occurred(&self) {
                PersistenceNotifierGuard::optionally_notify(&self.total_consistency_lock, &self.persistence_notifier, || {
                        let mut should_persist = NotifyOption::SkipPersist;
                        if self.process_background_events() { should_persist = NotifyOption::DoPersist; }
 
-                       let mut channel_state_lock = self.channel_state.lock().unwrap();
-                       let channel_state = &mut *channel_state_lock;
-                       for (_, chan) in channel_state.by_id.iter_mut() {
-                               match chan.channel_update_status() {
-                                       ChannelUpdateStatus::Enabled if !chan.is_live() => chan.set_channel_update_status(ChannelUpdateStatus::DisabledStaged),
-                                       ChannelUpdateStatus::Disabled if chan.is_live() => chan.set_channel_update_status(ChannelUpdateStatus::EnabledStaged),
-                                       ChannelUpdateStatus::DisabledStaged if chan.is_live() => chan.set_channel_update_status(ChannelUpdateStatus::Enabled),
-                                       ChannelUpdateStatus::EnabledStaged if !chan.is_live() => chan.set_channel_update_status(ChannelUpdateStatus::Disabled),
-                                       ChannelUpdateStatus::DisabledStaged if !chan.is_live() => {
-                                               if let Ok(update) = self.get_channel_update_for_broadcast(&chan) {
-                                                       channel_state.pending_msg_events.push(events::MessageSendEvent::BroadcastChannelUpdate {
-                                                               msg: update
-                                                       });
-                                               }
-                                               should_persist = NotifyOption::DoPersist;
-                                               chan.set_channel_update_status(ChannelUpdateStatus::Disabled);
-                                       },
-                                       ChannelUpdateStatus::EnabledStaged if chan.is_live() => {
-                                               if let Ok(update) = self.get_channel_update_for_broadcast(&chan) {
-                                                       channel_state.pending_msg_events.push(events::MessageSendEvent::BroadcastChannelUpdate {
-                                                               msg: update
-                                                       });
-                                               }
-                                               should_persist = NotifyOption::DoPersist;
-                                               chan.set_channel_update_status(ChannelUpdateStatus::Enabled);
-                                       },
-                                       _ => {},
-                               }
+                       let new_feerate = self.fee_estimator.get_est_sat_per_1000_weight(ConfirmationTarget::Normal);
+
+                       let mut handle_errors = Vec::new();
+                       {
+                               let mut channel_state_lock = self.channel_state.lock().unwrap();
+                               let channel_state = &mut *channel_state_lock;
+                               let pending_msg_events = &mut channel_state.pending_msg_events;
+                               let short_to_id = &mut channel_state.short_to_id;
+                               channel_state.by_id.retain(|chan_id, chan| {
+                                       match chan.channel_update_status() {
+                                               ChannelUpdateStatus::Enabled if !chan.is_live() => chan.set_channel_update_status(ChannelUpdateStatus::DisabledStaged),
+                                               ChannelUpdateStatus::Disabled if chan.is_live() => chan.set_channel_update_status(ChannelUpdateStatus::EnabledStaged),
+                                               ChannelUpdateStatus::DisabledStaged if chan.is_live() => chan.set_channel_update_status(ChannelUpdateStatus::Enabled),
+                                               ChannelUpdateStatus::EnabledStaged if !chan.is_live() => chan.set_channel_update_status(ChannelUpdateStatus::Disabled),
+                                               ChannelUpdateStatus::DisabledStaged if !chan.is_live() => {
+                                                       if let Ok(update) = self.get_channel_update_for_broadcast(&chan) {
+                                                               pending_msg_events.push(events::MessageSendEvent::BroadcastChannelUpdate {
+                                                                       msg: update
+                                                               });
+                                                       }
+                                                       should_persist = NotifyOption::DoPersist;
+                                                       chan.set_channel_update_status(ChannelUpdateStatus::Disabled);
+                                               },
+                                               ChannelUpdateStatus::EnabledStaged if chan.is_live() => {
+                                                       if let Ok(update) = self.get_channel_update_for_broadcast(&chan) {
+                                                               pending_msg_events.push(events::MessageSendEvent::BroadcastChannelUpdate {
+                                                                       msg: update
+                                                               });
+                                                       }
+                                                       should_persist = NotifyOption::DoPersist;
+                                                       chan.set_channel_update_status(ChannelUpdateStatus::Enabled);
+                                               },
+                                               _ => {},
+                                       }
+
+                                       let counterparty_node_id = chan.get_counterparty_node_id();
+                                       let (retain_channel, chan_needs_persist, err) = self.update_channel_fee(short_to_id, pending_msg_events, chan_id, chan, new_feerate);
+                                       if chan_needs_persist == NotifyOption::DoPersist { should_persist = NotifyOption::DoPersist; }
+                                       if err.is_err() {
+                                               handle_errors.push((err, counterparty_node_id));
+                                       }
+                                       retain_channel
+                               });
+                       }
+
+                       for (err, counterparty_node_id) in handle_errors.drain(..) {
+                               let _ = handle_error!(self, err, counterparty_node_id);
                        }
 
                        should_persist
@@ -2786,16 +2961,22 @@ impl<Signer: Sign, M: Deref, T: Deref, K: Deref, F: Deref, L: Deref> ChannelMana
                                                                         HTLCFailReason::Reason { failure_code: 0x4000|15, data: htlc_msat_height_data });
                                } else {
                                        match self.claim_funds_from_hop(channel_state.as_mut().unwrap(), htlc.prev_hop, payment_preimage) {
-                                               Err(Some(e)) => {
-                                                       if let msgs::ErrorAction::IgnoreError = e.1.err.action {
+                                               ClaimFundsFromHop::MonitorUpdateFail(pk, err, _) => {
+                                                       if let msgs::ErrorAction::IgnoreError = err.err.action {
                                                                // We got a temporary failure updating monitor, but will claim the
                                                                // HTLC when the monitor updating is restored (or on chain).
-                                                               log_error!(self.logger, "Temporary failure claiming HTLC, treating as success: {}", e.1.err.err);
+                                                               log_error!(self.logger, "Temporary failure claiming HTLC, treating as success: {}", err.err.err);
                                                                claimed_any_htlcs = true;
-                                                       } else { errs.push(e); }
+                                                       } else { errs.push((pk, err)); }
+                                               },
+                                               ClaimFundsFromHop::PrevHopForceClosed => unreachable!("We already checked for channel existence, we can't fail here!"),
+                                               ClaimFundsFromHop::DuplicateClaim => {
+                                                       // While we should never get here in most cases, if we do, it likely
+                                                       // indicates that the HTLC was timed out some time ago and is no longer
+                                                       // available to be claimed. Thus, it does not make sense to set
+                                                       // `claimed_any_htlcs`.
                                                },
-                                               Err(None) => unreachable!("We already checked for channel existence, we can't fail here!"),
-                                               Ok(()) => claimed_any_htlcs = true,
+                                               ClaimFundsFromHop::Success(_) => claimed_any_htlcs = true,
                                        }
                                }
                        }
@@ -2813,62 +2994,68 @@ impl<Signer: Sign, M: Deref, T: Deref, K: Deref, F: Deref, L: Deref> ChannelMana
                } else { false }
        }
 
-       fn claim_funds_from_hop(&self, channel_state_lock: &mut MutexGuard<ChannelHolder<Signer>>, prev_hop: HTLCPreviousHopData, payment_preimage: PaymentPreimage) -> Result<(), Option<(PublicKey, MsgHandleErrInternal)>> {
+       fn claim_funds_from_hop(&self, channel_state_lock: &mut MutexGuard<ChannelHolder<Signer>>, prev_hop: HTLCPreviousHopData, payment_preimage: PaymentPreimage) -> ClaimFundsFromHop {
                //TODO: Delay the claimed_funds relaying just like we do outbound relay!
                let channel_state = &mut **channel_state_lock;
                let chan_id = match channel_state.short_to_id.get(&prev_hop.short_channel_id) {
                        Some(chan_id) => chan_id.clone(),
                        None => {
-                               return Err(None)
+                               return ClaimFundsFromHop::PrevHopForceClosed
                        }
                };
 
                if let hash_map::Entry::Occupied(mut chan) = channel_state.by_id.entry(chan_id) {
-                       let was_frozen_for_monitor = chan.get().is_awaiting_monitor_update();
                        match chan.get_mut().get_update_fulfill_htlc_and_commit(prev_hop.htlc_id, payment_preimage, &self.logger) {
-                               Ok((msgs, monitor_option)) => {
-                                       if let Some(monitor_update) = monitor_option {
+                               Ok(msgs_monitor_option) => {
+                                       if let UpdateFulfillCommitFetch::NewClaim { msgs, htlc_value_msat, monitor_update } = msgs_monitor_option {
                                                if let Err(e) = self.chain_monitor.update_channel(chan.get().get_funding_txo().unwrap(), monitor_update) {
-                                                       if was_frozen_for_monitor {
-                                                               assert!(msgs.is_none());
-                                                       } else {
-                                                               return Err(Some((chan.get().get_counterparty_node_id(), handle_monitor_err!(self, e, channel_state, chan, RAACommitmentOrder::CommitmentFirst, false, msgs.is_some()).unwrap_err())));
-                                                       }
+                                                       log_given_level!(self.logger, if e == ChannelMonitorUpdateErr::PermanentFailure { Level::Error } else { Level::Debug },
+                                                               "Failed to update channel monitor with preimage {:?}: {:?}",
+                                                               payment_preimage, e);
+                                                       return ClaimFundsFromHop::MonitorUpdateFail(
+                                                               chan.get().get_counterparty_node_id(),
+                                                               handle_monitor_err!(self, e, channel_state, chan, RAACommitmentOrder::CommitmentFirst, false, msgs.is_some()).unwrap_err(),
+                                                               Some(htlc_value_msat)
+                                                       );
                                                }
+                                               if let Some((msg, commitment_signed)) = msgs {
+                                                       log_debug!(self.logger, "Claiming funds for HTLC with preimage {} resulted in a commitment_signed for channel {}",
+                                                               log_bytes!(payment_preimage.0), log_bytes!(chan.get().channel_id()));
+                                                       channel_state.pending_msg_events.push(events::MessageSendEvent::UpdateHTLCs {
+                                                               node_id: chan.get().get_counterparty_node_id(),
+                                                               updates: msgs::CommitmentUpdate {
+                                                                       update_add_htlcs: Vec::new(),
+                                                                       update_fulfill_htlcs: vec![msg],
+                                                                       update_fail_htlcs: Vec::new(),
+                                                                       update_fail_malformed_htlcs: Vec::new(),
+                                                                       update_fee: None,
+                                                                       commitment_signed,
+                                                               }
+                                                       });
+                                               }
+                                               return ClaimFundsFromHop::Success(htlc_value_msat);
+                                       } else {
+                                               return ClaimFundsFromHop::DuplicateClaim;
                                        }
-                                       if let Some((msg, commitment_signed)) = msgs {
-                                               log_debug!(self.logger, "Claiming funds for HTLC with preimage {} resulted in a commitment_signed for channel {}",
-                                                       log_bytes!(payment_preimage.0), log_bytes!(chan.get().channel_id()));
-                                               channel_state.pending_msg_events.push(events::MessageSendEvent::UpdateHTLCs {
-                                                       node_id: chan.get().get_counterparty_node_id(),
-                                                       updates: msgs::CommitmentUpdate {
-                                                               update_add_htlcs: Vec::new(),
-                                                               update_fulfill_htlcs: vec![msg],
-                                                               update_fail_htlcs: Vec::new(),
-                                                               update_fail_malformed_htlcs: Vec::new(),
-                                                               update_fee: None,
-                                                               commitment_signed,
-                                                       }
-                                               });
-                                       }
-                                       return Ok(())
                                },
-                               Err(e) => {
-                                       // TODO: Do something with e?
-                                       // This should only occur if we are claiming an HTLC at the same time as the
-                                       // HTLC is being failed (eg because a block is being connected and this caused
-                                       // an HTLC to time out). This should, of course, only occur if the user is the
-                                       // one doing the claiming (as it being a part of a peer claim would imply we're
-                                       // about to lose funds) and only if the lock in claim_funds was dropped as a
-                                       // previous HTLC was failed (thus not for an MPP payment).
-                                       debug_assert!(false, "This shouldn't be reachable except in absurdly rare cases between monitor updates and HTLC timeouts: {:?}", e);
-                                       return Err(None)
+                               Err((e, monitor_update)) => {
+                                       if let Err(e) = self.chain_monitor.update_channel(chan.get().get_funding_txo().unwrap(), monitor_update) {
+                                               log_given_level!(self.logger, if e == ChannelMonitorUpdateErr::PermanentFailure { Level::Error } else { Level::Info },
+                                                       "Failed to update channel monitor with preimage {:?} immediately prior to force-close: {:?}",
+                                                       payment_preimage, e);
+                                       }
+                                       let counterparty_node_id = chan.get().get_counterparty_node_id();
+                                       let (drop, res) = convert_chan_err!(self, e, channel_state.short_to_id, chan.get_mut(), &chan_id);
+                                       if drop {
+                                               chan.remove_entry();
+                                       }
+                                       return ClaimFundsFromHop::MonitorUpdateFail(counterparty_node_id, res, None);
                                },
                        }
                } else { unreachable!(); }
        }
 
-       fn claim_funds_internal(&self, mut channel_state_lock: MutexGuard<ChannelHolder<Signer>>, source: HTLCSource, payment_preimage: PaymentPreimage) {
+       fn claim_funds_internal(&self, mut channel_state_lock: MutexGuard<ChannelHolder<Signer>>, source: HTLCSource, payment_preimage: PaymentPreimage, forwarded_htlc_value_msat: Option<u64>, from_onchain: bool) {
                match source {
                        HTLCSource::OutboundRoute { session_priv, .. } => {
                                mem::drop(channel_state_lock);
@@ -2887,29 +3074,51 @@ impl<Signer: Sign, M: Deref, T: Deref, K: Deref, F: Deref, L: Deref> ChannelMana
                        },
                        HTLCSource::PreviousHopData(hop_data) => {
                                let prev_outpoint = hop_data.outpoint;
-                               if let Err((counterparty_node_id, err)) = match self.claim_funds_from_hop(&mut channel_state_lock, hop_data, payment_preimage) {
-                                       Ok(()) => Ok(()),
-                                       Err(None) => {
-                                               let preimage_update = ChannelMonitorUpdate {
-                                                       update_id: CLOSED_CHANNEL_UPDATE_ID,
-                                                       updates: vec![ChannelMonitorUpdateStep::PaymentPreimage {
-                                                               payment_preimage: payment_preimage.clone(),
-                                                       }],
-                                               };
-                                               // We update the ChannelMonitor on the backward link, after
-                                               // receiving an offchain preimage event from the forward link (the
-                                               // event being update_fulfill_htlc).
-                                               if let Err(e) = self.chain_monitor.update_channel(prev_outpoint, preimage_update) {
-                                                       log_error!(self.logger, "Critical error: failed to update channel monitor with preimage {:?}: {:?}",
-                                                                  payment_preimage, e);
-                                               }
-                                               Ok(())
-                                       },
-                                       Err(Some(res)) => Err(res),
-                               } {
-                                       mem::drop(channel_state_lock);
-                                       let res: Result<(), _> = Err(err);
-                                       let _ = handle_error!(self, res, counterparty_node_id);
+                               let res = self.claim_funds_from_hop(&mut channel_state_lock, hop_data, payment_preimage);
+                               let claimed_htlc = if let ClaimFundsFromHop::DuplicateClaim = res { false } else { true };
+                               let htlc_claim_value_msat = match res {
+                                       ClaimFundsFromHop::MonitorUpdateFail(_, _, amt_opt) => amt_opt,
+                                       ClaimFundsFromHop::Success(amt) => Some(amt),
+                                       _ => None,
+                               };
+                               if let ClaimFundsFromHop::PrevHopForceClosed = res {
+                                       let preimage_update = ChannelMonitorUpdate {
+                                               update_id: CLOSED_CHANNEL_UPDATE_ID,
+                                               updates: vec![ChannelMonitorUpdateStep::PaymentPreimage {
+                                                       payment_preimage: payment_preimage.clone(),
+                                               }],
+                                       };
+                                       // We update the ChannelMonitor on the backward link, after
+                                       // receiving an offchain preimage event from the forward link (the
+                                       // event being update_fulfill_htlc).
+                                       if let Err(e) = self.chain_monitor.update_channel(prev_outpoint, preimage_update) {
+                                               log_error!(self.logger, "Critical error: failed to update channel monitor with preimage {:?}: {:?}",
+                                                                                        payment_preimage, e);
+                                       }
+                                       // Note that we do *not* set `claimed_htlc` to false here. In fact, this
+                                       // totally could be a duplicate claim, but we have no way of knowing
+                                       // without interrogating the `ChannelMonitor` we've provided the above
+                                       // update to. Instead, we simply document in `PaymentForwarded` that this
+                                       // can happen.
+                               }
+                               mem::drop(channel_state_lock);
+                               if let ClaimFundsFromHop::MonitorUpdateFail(pk, err, _) = res {
+                                       let result: Result<(), _> = Err(err);
+                                       let _ = handle_error!(self, result, pk);
+                               }
+
+                               if claimed_htlc {
+                                       if let Some(forwarded_htlc_value) = forwarded_htlc_value_msat {
+                                               let fee_earned_msat = if let Some(claimed_htlc_value) = htlc_claim_value_msat {
+                                                       Some(claimed_htlc_value - forwarded_htlc_value)
+                                               } else { None };
+
+                                               let mut pending_events = self.pending_events.lock().unwrap();
+                                               pending_events.push(events::Event::PaymentForwarded {
+                                                       fee_earned_msat,
+                                                       claim_from_onchain_tx: from_onchain,
+                                               });
+                                       }
                                }
                        },
                }
@@ -2983,7 +3192,7 @@ impl<Signer: Sign, M: Deref, T: Deref, K: Deref, F: Deref, L: Deref> ChannelMana
                        return Err(MsgHandleErrInternal::send_err_msg_no_close("Unknown genesis block hash".to_owned(), msg.temporary_channel_id.clone()));
                }
 
-               let channel = Channel::new_from_req(&self.fee_estimator, &self.keys_manager, counterparty_node_id.clone(), their_features, msg, 0, &self.default_configuration)
+               let channel = Channel::new_from_req(&self.fee_estimator, &self.keys_manager, counterparty_node_id.clone(), &their_features, msg, 0, &self.default_configuration)
                        .map_err(|e| MsgHandleErrInternal::from_chan_no_close(e, msg.temporary_channel_id))?;
                let mut channel_state_lock = self.channel_state.lock().unwrap();
                let channel_state = &mut *channel_state_lock;
@@ -3009,7 +3218,7 @@ impl<Signer: Sign, M: Deref, T: Deref, K: Deref, F: Deref, L: Deref> ChannelMana
                                        if chan.get().get_counterparty_node_id() != *counterparty_node_id {
                                                return Err(MsgHandleErrInternal::send_err_msg_no_close("Got a message for a channel from the wrong node!".to_owned(), msg.temporary_channel_id));
                                        }
-                                       try_chan_entry!(self, chan.get_mut().accept_channel(&msg, &self.default_configuration, their_features), channel_state, chan);
+                                       try_chan_entry!(self, chan.get_mut().accept_channel(&msg, &self.default_configuration, &their_features), channel_state, chan);
                                        (chan.get().get_value_satoshis(), chan.get().get_funding_redeemscript().to_v0_p2wsh(), chan.get().get_user_id())
                                },
                                hash_map::Entry::Vacant(_) => return Err(MsgHandleErrInternal::send_err_msg_no_close("Failed to find corresponding channel".to_owned(), msg.temporary_channel_id))
@@ -3146,7 +3355,8 @@ impl<Signer: Sign, M: Deref, T: Deref, K: Deref, F: Deref, L: Deref> ChannelMana
        }
 
        fn internal_shutdown(&self, counterparty_node_id: &PublicKey, their_features: &InitFeatures, msg: &msgs::Shutdown) -> Result<(), MsgHandleErrInternal> {
-               let (mut dropped_htlcs, chan_option) = {
+               let mut dropped_htlcs: Vec<(HTLCSource, PaymentHash)>;
+               let result: Result<(), _> = loop {
                        let mut channel_state_lock = self.channel_state.lock().unwrap();
                        let channel_state = &mut *channel_state_lock;
 
@@ -3155,25 +3365,43 @@ impl<Signer: Sign, M: Deref, T: Deref, K: Deref, F: Deref, L: Deref> ChannelMana
                                        if chan_entry.get().get_counterparty_node_id() != *counterparty_node_id {
                                                return Err(MsgHandleErrInternal::send_err_msg_no_close("Got a message for a channel from the wrong node!".to_owned(), msg.channel_id));
                                        }
-                                       let (shutdown, closing_signed, dropped_htlcs) = try_chan_entry!(self, chan_entry.get_mut().shutdown(&self.fee_estimator, &their_features, &msg), channel_state, chan_entry);
+
+                                       if !chan_entry.get().received_shutdown() {
+                                               log_info!(self.logger, "Received a shutdown message from our couterparty for channel {}{}.",
+                                                       log_bytes!(msg.channel_id),
+                                                       if chan_entry.get().sent_shutdown() { " after we initiated shutdown" } else { "" });
+                                       }
+
+                                       let (shutdown, closing_signed, monitor_update, htlcs) = try_chan_entry!(self, chan_entry.get_mut().shutdown(&self.fee_estimator, &self.keys_manager, &their_features, &msg), channel_state, chan_entry);
+                                       dropped_htlcs = htlcs;
+
+                                       // Update the monitor with the shutdown script if necessary.
+                                       if let Some(monitor_update) = monitor_update {
+                                               if let Err(e) = self.chain_monitor.update_channel(chan_entry.get().get_funding_txo().unwrap(), monitor_update) {
+                                                       let (result, is_permanent) =
+                                                               handle_monitor_err!(self, e, channel_state.short_to_id, chan_entry.get_mut(), RAACommitmentOrder::CommitmentFirst, false, false, Vec::new(), Vec::new(), chan_entry.key());
+                                                       if is_permanent {
+                                                               remove_channel!(channel_state, chan_entry);
+                                                               break result;
+                                                       }
+                                               }
+                                       }
+
                                        if let Some(msg) = shutdown {
                                                channel_state.pending_msg_events.push(events::MessageSendEvent::SendShutdown {
-                                                       node_id: counterparty_node_id.clone(),
+                                                       node_id: *counterparty_node_id,
                                                        msg,
                                                });
                                        }
                                        if let Some(msg) = closing_signed {
+                                               // TODO: Do not send this if the monitor update failed.
                                                channel_state.pending_msg_events.push(events::MessageSendEvent::SendClosingSigned {
-                                                       node_id: counterparty_node_id.clone(),
+                                                       node_id: *counterparty_node_id,
                                                        msg,
                                                });
                                        }
-                                       if chan_entry.get().is_shutdown() {
-                                               if let Some(short_id) = chan_entry.get().get_short_channel_id() {
-                                                       channel_state.short_to_id.remove(&short_id);
-                                               }
-                                               (dropped_htlcs, Some(chan_entry.remove_entry().1))
-                                       } else { (dropped_htlcs, None) }
+
+                                       break Ok(());
                                },
                                hash_map::Entry::Vacant(_) => return Err(MsgHandleErrInternal::send_err_msg_no_close("Failed to find corresponding channel".to_owned(), msg.channel_id))
                        }
@@ -3181,14 +3409,8 @@ impl<Signer: Sign, M: Deref, T: Deref, K: Deref, F: Deref, L: Deref> ChannelMana
                for htlc_source in dropped_htlcs.drain(..) {
                        self.fail_htlc_backwards_internal(self.channel_state.lock().unwrap(), htlc_source.0, &htlc_source.1, HTLCFailReason::Reason { failure_code: 0x4000 | 8, data: Vec::new() });
                }
-               if let Some(chan) = chan_option {
-                       if let Ok(update) = self.get_channel_update_for_broadcast(&chan) {
-                               let mut channel_state = self.channel_state.lock().unwrap();
-                               channel_state.pending_msg_events.push(events::MessageSendEvent::BroadcastChannelUpdate {
-                                       msg: update
-                               });
-                       }
-               }
+
+               let _ = handle_error!(self, result, *counterparty_node_id);
                Ok(())
        }
 
@@ -3305,7 +3527,7 @@ impl<Signer: Sign, M: Deref, T: Deref, K: Deref, F: Deref, L: Deref> ChannelMana
 
        fn internal_update_fulfill_htlc(&self, counterparty_node_id: &PublicKey, msg: &msgs::UpdateFulfillHTLC) -> Result<(), MsgHandleErrInternal> {
                let mut channel_lock = self.channel_state.lock().unwrap();
-               let htlc_source = {
+               let (htlc_source, forwarded_htlc_value) = {
                        let channel_state = &mut *channel_lock;
                        match channel_state.by_id.entry(msg.channel_id) {
                                hash_map::Entry::Occupied(mut chan) => {
@@ -3317,7 +3539,7 @@ impl<Signer: Sign, M: Deref, T: Deref, K: Deref, F: Deref, L: Deref> ChannelMana
                                hash_map::Entry::Vacant(_) => return Err(MsgHandleErrInternal::send_err_msg_no_close("Failed to find corresponding channel".to_owned(), msg.channel_id))
                        }
                };
-               self.claim_funds_internal(channel_lock, htlc_source, msg.payment_preimage.clone());
+               self.claim_funds_internal(channel_lock, htlc_source, msg.payment_preimage.clone(), Some(forwarded_htlc_value), false);
                Ok(())
        }
 
@@ -3625,73 +3847,17 @@ impl<Signer: Sign, M: Deref, T: Deref, K: Deref, F: Deref, L: Deref> ChannelMana
                Ok(())
        }
 
-       /// Begin Update fee process. Allowed only on an outbound channel.
-       /// If successful, will generate a UpdateHTLCs event, so you should probably poll
-       /// PeerManager::process_events afterwards.
-       /// Note: This API is likely to change!
-       /// (C-not exported) Cause its doc(hidden) anyway
-       #[doc(hidden)]
-       pub fn update_fee(&self, channel_id: [u8;32], feerate_per_kw: u32) -> Result<(), APIError> {
-               let _persistence_guard = PersistenceNotifierGuard::notify_on_drop(&self.total_consistency_lock, &self.persistence_notifier);
-               let counterparty_node_id;
-               let err: Result<(), _> = loop {
-                       let mut channel_state_lock = self.channel_state.lock().unwrap();
-                       let channel_state = &mut *channel_state_lock;
-
-                       match channel_state.by_id.entry(channel_id) {
-                               hash_map::Entry::Vacant(_) => return Err(APIError::APIMisuseError{err: format!("Failed to find corresponding channel for id {}", channel_id.to_hex())}),
-                               hash_map::Entry::Occupied(mut chan) => {
-                                       if !chan.get().is_outbound() {
-                                               return Err(APIError::APIMisuseError{err: "update_fee cannot be sent for an inbound channel".to_owned()});
-                                       }
-                                       if chan.get().is_awaiting_monitor_update() {
-                                               return Err(APIError::MonitorUpdateFailed);
-                                       }
-                                       if !chan.get().is_live() {
-                                               return Err(APIError::ChannelUnavailable{err: "Channel is either not yet fully established or peer is currently disconnected".to_owned()});
-                                       }
-                                       counterparty_node_id = chan.get().get_counterparty_node_id();
-                                       if let Some((update_fee, commitment_signed, monitor_update)) =
-                                                       break_chan_entry!(self, chan.get_mut().send_update_fee_and_commit(feerate_per_kw, &self.logger), channel_state, chan)
-                                       {
-                                               if let Err(_e) = self.chain_monitor.update_channel(chan.get().get_funding_txo().unwrap(), monitor_update) {
-                                                       unimplemented!();
-                                               }
-                                               log_debug!(self.logger, "Updating fee resulted in a commitment_signed for channel {}", log_bytes!(chan.get().channel_id()));
-                                               channel_state.pending_msg_events.push(events::MessageSendEvent::UpdateHTLCs {
-                                                       node_id: chan.get().get_counterparty_node_id(),
-                                                       updates: msgs::CommitmentUpdate {
-                                                               update_add_htlcs: Vec::new(),
-                                                               update_fulfill_htlcs: Vec::new(),
-                                                               update_fail_htlcs: Vec::new(),
-                                                               update_fail_malformed_htlcs: Vec::new(),
-                                                               update_fee: Some(update_fee),
-                                                               commitment_signed,
-                                                       },
-                                               });
-                                       }
-                               },
-                       }
-                       return Ok(())
-               };
-
-               match handle_error!(self, err, counterparty_node_id) {
-                       Ok(_) => unreachable!(),
-                       Err(e) => { Err(APIError::APIMisuseError { err: e.err })}
-               }
-       }
-
        /// Process pending events from the `chain::Watch`, returning whether any events were processed.
        fn process_pending_monitor_events(&self) -> bool {
                let mut failed_channels = Vec::new();
-               let pending_monitor_events = self.chain_monitor.release_pending_monitor_events();
+               let mut pending_monitor_events = self.chain_monitor.release_pending_monitor_events();
                let has_pending_monitor_events = !pending_monitor_events.is_empty();
-               for monitor_event in pending_monitor_events {
+               for monitor_event in pending_monitor_events.drain(..) {
                        match monitor_event {
                                MonitorEvent::HTLCEvent(htlc_update) => {
                                        if let Some(preimage) = htlc_update.payment_preimage {
                                                log_trace!(self.logger, "Claiming HTLC with preimage {} from our monitor", log_bytes!(preimage.0));
-                                               self.claim_funds_internal(self.channel_state.lock().unwrap(), htlc_update.source, preimage);
+                                               self.claim_funds_internal(self.channel_state.lock().unwrap(), htlc_update.source, preimage, htlc_update.onchain_value_satoshis.map(|v| v * 1000), true);
                                        } else {
                                                log_trace!(self.logger, "Failing HTLC with hash {} from our monitor", log_bytes!(htlc_update.payment_hash.0));
                                                self.fail_htlc_backwards_internal(self.channel_state.lock().unwrap(), htlc_update.source, &htlc_update.payment_hash, HTLCFailReason::Reason { failure_code: 0x4000 | 8, data: Vec::new() });
@@ -3986,7 +4152,7 @@ where
                                result = NotifyOption::DoPersist;
                        }
 
-                       let mut pending_events = std::mem::replace(&mut *self.pending_events.lock().unwrap(), vec![]);
+                       let mut pending_events = mem::replace(&mut *self.pending_events.lock().unwrap(), vec![]);
                        if !pending_events.is_empty() {
                                result = NotifyOption::DoPersist;
                        }
@@ -4606,7 +4772,7 @@ impl_writeable_tlv_based!(HTLCPreviousHopData, {
 });
 
 impl Writeable for ClaimableHTLC {
-       fn write<W: Writer>(&self, writer: &mut W) -> Result<(), ::std::io::Error> {
+       fn write<W: Writer>(&self, writer: &mut W) -> Result<(), io::Error> {
                let payment_data = match &self.onion_payload {
                        OnionPayload::Invoice(data) => Some(data.clone()),
                        _ => None,
@@ -4710,7 +4876,7 @@ impl<Signer: Sign, M: Deref, T: Deref, K: Deref, F: Deref, L: Deref> Writeable f
         F::Target: FeeEstimator,
         L::Target: Logger,
 {
-       fn write<W: Writer>(&self, writer: &mut W) -> Result<(), ::std::io::Error> {
+       fn write<W: Writer>(&self, writer: &mut W) -> Result<(), io::Error> {
                let _consistency_lock = self.total_consistency_lock.write().unwrap();
 
                write_ver_prefix!(writer, SERIALIZATION_VERSION, MIN_SERIALIZATION_VERSION);
@@ -4906,7 +5072,7 @@ impl<'a, Signer: Sign, M: Deref, T: Deref, K: Deref, F: Deref, L: Deref>
         F::Target: FeeEstimator,
         L::Target: Logger,
 {
-       fn read<R: ::std::io::Read>(reader: &mut R, args: ChannelManagerReadArgs<'a, Signer, M, T, K, F, L>) -> Result<Self, DecodeError> {
+       fn read<R: io::Read>(reader: &mut R, args: ChannelManagerReadArgs<'a, Signer, M, T, K, F, L>) -> Result<Self, DecodeError> {
                let (blockhash, chan_manager) = <(BlockHash, ChannelManager<Signer, M, T, K, F, L>)>::read(reader, args)?;
                Ok((blockhash, Arc::new(chan_manager)))
        }
@@ -4920,7 +5086,7 @@ impl<'a, Signer: Sign, M: Deref, T: Deref, K: Deref, F: Deref, L: Deref>
         F::Target: FeeEstimator,
         L::Target: Logger,
 {
-       fn read<R: ::std::io::Read>(reader: &mut R, mut args: ChannelManagerReadArgs<'a, Signer, M, T, K, F, L>) -> Result<Self, DecodeError> {
+       fn read<R: io::Read>(reader: &mut R, mut args: ChannelManagerReadArgs<'a, Signer, M, T, K, F, L>) -> Result<Self, DecodeError> {
                let _ver = read_ver_prefix!(reader, SERIALIZATION_VERSION);
 
                let genesis_hash: BlockHash = Readable::read(reader)?;
@@ -4956,6 +5122,10 @@ impl<'a, Signer: Sign, M: Deref, T: Deref, K: Deref, F: Deref, L: Deref>
                                                channel.get_cur_counterparty_commitment_transaction_number() > monitor.get_cur_counterparty_commitment_number() ||
                                                channel.get_latest_monitor_update_id() < monitor.get_latest_update_id() {
                                        // But if the channel is behind of the monitor, close the channel:
+                                       log_error!(args.logger, "A ChannelManager is stale compared to the current ChannelMonitor!");
+                                       log_error!(args.logger, " The channel will be force-closed and the latest commitment transaction from the ChannelMonitor broadcast.");
+                                       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());
                                        let (_, mut new_failed_htlcs) = channel.force_shutdown(true);
                                        failed_htlcs.append(&mut new_failed_htlcs);
                                        monitor.broadcast_latest_holder_commitment_txn(&args.tx_broadcaster, &args.logger);
@@ -5110,23 +5280,26 @@ impl<'a, Signer: Sign, M: Deref, T: Deref, K: Deref, F: Deref, L: Deref>
 mod tests {
        use bitcoin::hashes::Hash;
        use bitcoin::hashes::sha256::Hash as Sha256;
-       use core::sync::atomic::{AtomicBool, Ordering};
        use core::time::Duration;
        use ln::{PaymentPreimage, PaymentHash, PaymentSecret};
-       use ln::channelmanager::PersistenceNotifier;
+       use ln::channelmanager::PaymentSendFailure;
        use ln::features::{InitFeatures, InvoiceFeatures};
        use ln::functional_test_utils::*;
        use ln::msgs;
        use ln::msgs::ChannelMessageHandler;
        use routing::router::{get_keysend_route, get_route};
+       use util::errors::APIError;
        use util::events::{Event, MessageSendEvent, MessageSendEventsProvider};
        use util::test_utils;
-       use std::sync::Arc;
-       use std::thread;
 
        #[cfg(feature = "std")]
        #[test]
        fn test_wait_timeout() {
+               use ln::channelmanager::PersistenceNotifier;
+               use sync::Arc;
+               use core::sync::atomic::{AtomicBool, Ordering};
+               use std::thread;
+
                let persistence_notifier = Arc::new(PersistenceNotifier::new());
                let thread_notifier = Arc::clone(&persistence_notifier);
 
@@ -5176,6 +5349,12 @@ mod tests {
                let node_chanmgrs = create_node_chanmgrs(3, &node_cfgs, &[None, None, None]);
                let nodes = create_network(3, &node_cfgs, &node_chanmgrs);
 
+               // All nodes start with a persistable update pending as `create_network` connects each node
+               // with all other nodes to make most tests simpler.
+               assert!(nodes[0].node.await_persistable_update_timeout(Duration::from_millis(1)));
+               assert!(nodes[1].node.await_persistable_update_timeout(Duration::from_millis(1)));
+               assert!(nodes[2].node.await_persistable_update_timeout(Duration::from_millis(1)));
+
                let mut chan = create_announced_chan_between_nodes(&nodes, 0, 1, InitFeatures::known(), InitFeatures::known());
 
                // We check that the channel info nodes have doesn't change too early, even though we try
@@ -5497,6 +5676,39 @@ mod tests {
 
                nodes[1].logger.assert_log_contains("lightning::ln::channelmanager".to_string(), "We don't support MPP keysend payments".to_string(), 1);
        }
+
+       #[test]
+       fn test_multi_hop_missing_secret() {
+               let chanmon_cfgs = create_chanmon_cfgs(4);
+               let node_cfgs = create_node_cfgs(4, &chanmon_cfgs);
+               let node_chanmgrs = create_node_chanmgrs(4, &node_cfgs, &[None, None, None, None]);
+               let nodes = create_network(4, &node_cfgs, &node_chanmgrs);
+
+               let chan_1_id = create_announced_chan_between_nodes(&nodes, 0, 1, InitFeatures::known(), InitFeatures::known()).0.contents.short_channel_id;
+               let chan_2_id = create_announced_chan_between_nodes(&nodes, 0, 2, InitFeatures::known(), InitFeatures::known()).0.contents.short_channel_id;
+               let chan_3_id = create_announced_chan_between_nodes(&nodes, 1, 3, InitFeatures::known(), InitFeatures::known()).0.contents.short_channel_id;
+               let chan_4_id = create_announced_chan_between_nodes(&nodes, 2, 3, InitFeatures::known(), InitFeatures::known()).0.contents.short_channel_id;
+               let logger = test_utils::TestLogger::new();
+
+               // Marshall an MPP route.
+               let (_, payment_hash, _) = get_payment_preimage_hash!(&nodes[3]);
+               let net_graph_msg_handler = &nodes[0].net_graph_msg_handler;
+               let mut route = get_route(&nodes[0].node.get_our_node_id(), &net_graph_msg_handler.network_graph.read().unwrap(), &nodes[3].node.get_our_node_id(), Some(InvoiceFeatures::known()), None, &[], 100000, TEST_FINAL_CLTV, &logger).unwrap();
+               let path = route.paths[0].clone();
+               route.paths.push(path);
+               route.paths[0][0].pubkey = nodes[1].node.get_our_node_id();
+               route.paths[0][0].short_channel_id = chan_1_id;
+               route.paths[0][1].short_channel_id = chan_3_id;
+               route.paths[1][0].pubkey = nodes[2].node.get_our_node_id();
+               route.paths[1][0].short_channel_id = chan_2_id;
+               route.paths[1][1].short_channel_id = chan_4_id;
+
+               match nodes[0].node.send_payment(&route, payment_hash, &None).unwrap_err() {
+                       PaymentSendFailure::ParameterError(APIError::APIMisuseError { ref err }) => {
+                               assert!(regex::Regex::new(r"Payment secret is required for multi-path payments").unwrap().is_match(err))                        },
+                       _ => panic!("unexpected error")
+               }
+       }
 }
 
 #[cfg(all(any(test, feature = "_test_utils"), feature = "unstable"))]
@@ -5508,7 +5720,7 @@ pub mod bench {
        use ln::channelmanager::{BestBlock, ChainParameters, ChannelManager, PaymentHash, PaymentPreimage};
        use ln::features::{InitFeatures, InvoiceFeatures};
        use ln::functional_test_utils::*;
-       use ln::msgs::ChannelMessageHandler;
+       use ln::msgs::{ChannelMessageHandler, Init};
        use routing::network_graph::NetworkGraph;
        use routing::router::get_route;
        use util::test_utils;
@@ -5571,6 +5783,8 @@ pub mod bench {
                });
                let node_b_holder = NodeHolder { node: &node_b };
 
+               node_a.peer_connected(&node_b.get_our_node_id(), &Init { features: InitFeatures::known() });
+               node_b.peer_connected(&node_a.get_our_node_id(), &Init { features: InitFeatures::known() });
                node_a.create_channel(node_b.get_our_node_id(), 8_000_000, 100_000_000, 42, None).unwrap();
                node_b.handle_open_channel(&node_a.get_our_node_id(), InitFeatures::known(), &get_event_msg!(node_a_holder, MessageSendEvent::SendOpenChannel, node_b.get_our_node_id()));
                node_a.handle_accept_channel(&node_b.get_our_node_id(), InitFeatures::known(), &get_event_msg!(node_b_holder, MessageSendEvent::SendAcceptChannel, node_a.get_our_node_id()));