Keep track of the Init Features for every connected/channel'd peer
[rust-lightning] / lightning / src / ln / channelmanager.rs
index 5a59d0c45af798aa22f0291b88eeb1b242dff52e..72193bafeb912144a628431bf03075cec8c1a384 100644 (file)
@@ -30,11 +30,11 @@ use chain::transaction::OutPoint;
 use ln::channel::{Channel, ChannelError};
 use ln::channelmonitor::{ChannelMonitor, ChannelMonitorUpdateErr, ManyChannelMonitor, CLTV_CLAIM_BUFFER, LATENCY_GRACE_PERIOD_BLOCKS, ANTI_REORG_DELAY};
 use ln::router::Route;
+use ln::features::InitFeatures;
 use ln::msgs;
-use ln::msgs::LocalFeatures;
 use ln::onion_utils;
 use ln::msgs::{ChannelMessageHandler, DecodeError, LightningError};
-use chain::keysinterface::KeysInterface;
+use chain::keysinterface::{ChannelKeys, KeysInterface};
 use util::config::UserConfig;
 use util::{byte_utils, events};
 use util::ser::{Readable, ReadableArgs, Writeable, Writer};
@@ -49,6 +49,8 @@ use std::sync::{Arc, Mutex, MutexGuard, RwLock};
 use std::sync::atomic::{AtomicUsize, Ordering};
 use std::time::Duration;
 
+const SIXTY_FIVE_ZEROS: [u8; 65] = [0; 65];
+
 // We hold various information about HTLC relay in the HTLC objects in Channel itself:
 //
 // Upon receipt of an HTLC from a peer, we'll give it a PendingHTLCStatus indicating if it should
@@ -254,8 +256,8 @@ pub(super) enum RAACommitmentOrder {
 }
 
 // Note this is only exposed in cfg(test):
-pub(super) struct ChannelHolder {
-       pub(super) by_id: HashMap<[u8; 32], Channel>,
+pub(super) struct ChannelHolder<ChanSigner: ChannelKeys> {
+       pub(super) by_id: HashMap<[u8; 32], Channel<ChanSigner>>,
        pub(super) short_to_id: HashMap<u64, [u8; 32]>,
        /// short channel id -> forward infos. Key of 0 means payments received
        /// Note that while this is held in the same mutex as the channels themselves, no consistency
@@ -272,23 +274,11 @@ pub(super) struct ChannelHolder {
        /// for broadcast messages, where ordering isn't as strict).
        pub(super) pending_msg_events: Vec<events::MessageSendEvent>,
 }
-pub(super) struct MutChannelHolder<'a> {
-       pub(super) by_id: &'a mut HashMap<[u8; 32], Channel>,
-       pub(super) short_to_id: &'a mut HashMap<u64, [u8; 32]>,
-       pub(super) forward_htlcs: &'a mut HashMap<u64, Vec<HTLCForwardInfo>>,
-       pub(super) claimable_htlcs: &'a mut HashMap<PaymentHash, Vec<(u64, HTLCPreviousHopData)>>,
-       pub(super) pending_msg_events: &'a mut Vec<events::MessageSendEvent>,
-}
-impl ChannelHolder {
-       pub(super) fn borrow_parts(&mut self) -> MutChannelHolder {
-               MutChannelHolder {
-                       by_id: &mut self.by_id,
-                       short_to_id: &mut self.short_to_id,
-                       forward_htlcs: &mut self.forward_htlcs,
-                       claimable_htlcs: &mut self.claimable_htlcs,
-                       pending_msg_events: &mut self.pending_msg_events,
-               }
-       }
+
+/// State we hold per-peer. In the future we should put channels in here, but for now we only hold
+/// the latest Init features we heard from the peer.
+struct PeerState {
+       latest_features: InitFeatures,
 }
 
 #[cfg(not(any(target_pointer_width = "32", target_pointer_width = "64")))]
@@ -324,11 +314,11 @@ const ERR: () = "You need at least 32 bit pointers (well, usize, but we'll assum
 /// spam due to quick disconnection/reconnection, updates are not sent until the channel has been
 /// offline for a full minute. In order to track this, you must call
 /// timer_chan_freshness_every_min roughly once per minute, though it doesn't have to be perfec.
-pub struct ChannelManager<'a> {
+pub struct ChannelManager<ChanSigner: ChannelKeys> {
        default_configuration: UserConfig,
        genesis_hash: Sha256dHash,
        fee_estimator: Arc<FeeEstimator>,
-       monitor: Arc<ManyChannelMonitor + 'a>,
+       monitor: Arc<ManyChannelMonitor>,
        tx_broadcaster: Arc<BroadcasterInterface>,
 
        #[cfg(test)]
@@ -339,18 +329,26 @@ pub struct ChannelManager<'a> {
        secp_ctx: Secp256k1<secp256k1::All>,
 
        #[cfg(test)]
-       pub(super) channel_state: Mutex<ChannelHolder>,
+       pub(super) channel_state: Mutex<ChannelHolder<ChanSigner>>,
        #[cfg(not(test))]
-       channel_state: Mutex<ChannelHolder>,
+       channel_state: Mutex<ChannelHolder<ChanSigner>>,
        our_network_key: SecretKey,
 
+       /// The bulk of our storage will eventually be here (channels and message queues and the like).
+       /// If we are connected to a peer we always at least have an entry here, even if no channels
+       /// are currently open with that peer.
+       /// 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.
+       per_peer_state: RwLock<HashMap<PublicKey, Mutex<PeerState>>>,
+
        pending_events: Mutex<Vec<events::Event>>,
        /// Used when we have to take a BIG lock to make sure everything is self-consistent.
        /// Essentially just when we're serializing ourselves out.
        /// Taken first everywhere where we are making changes before any other locks.
        total_consistency_lock: RwLock<()>,
 
-       keys_manager: Arc<KeysInterface>,
+       keys_manager: Arc<KeysInterface<ChanKeySigner = ChanSigner>>,
 
        logger: Arc<Logger>,
 }
@@ -427,19 +425,22 @@ pub struct ChannelDetails {
 }
 
 macro_rules! handle_error {
-       ($self: ident, $internal: expr) => {
+       ($self: ident, $internal: expr, $their_node_id: expr, $locked_channel_state: expr) => {
                match $internal {
                        Ok(msg) => Ok(msg),
                        Err(MsgHandleErrInternal { err, shutdown_finish }) => {
                                if let Some((shutdown_res, update_option)) = shutdown_finish {
                                        $self.finish_force_close_channel(shutdown_res);
                                        if let Some(update) = update_option {
-                                               let mut channel_state = $self.channel_state.lock().unwrap();
-                                               channel_state.pending_msg_events.push(events::MessageSendEvent::BroadcastChannelUpdate {
+                                               $locked_channel_state.pending_msg_events.push(events::MessageSendEvent::BroadcastChannelUpdate {
                                                        msg: update
                                                });
                                        }
                                }
+                               log_error!($self, "{}", err.err);
+                               if let msgs::ErrorAction::IgnoreError = err.action {
+                               } else { $locked_channel_state.pending_msg_events.push(events::MessageSendEvent::HandleError { node_id: $their_node_id, action: err.action.clone() }); }
+                               // Return error in case higher-API need one
                                Err(err)
                        },
                }
@@ -581,7 +582,7 @@ macro_rules! maybe_break_monitor_err {
        }
 }
 
-impl<'a> ChannelManager<'a> {
+impl<ChanSigner: ChannelKeys> ChannelManager<ChanSigner> {
        /// Constructs a new ChannelManager to hold several channels and route between them.
        ///
        /// This is the main "logic hub" for all channel-related actions, and implements
@@ -600,7 +601,7 @@ impl<'a> ChannelManager<'a> {
        /// the ChannelManager as a listener to the BlockNotifier and call the BlockNotifier's
        /// `block_(dis)connected` methods, which will notify all registered listeners in one
        /// go.
-       pub fn new(network: Network, feeest: Arc<FeeEstimator>, monitor: Arc<ManyChannelMonitor + 'a>, tx_broadcaster: Arc<BroadcasterInterface>, logger: Arc<Logger>,keys_manager: Arc<KeysInterface>, config: UserConfig, current_blockchain_height: usize) -> Result<Arc<ChannelManager<'a>>, secp256k1::Error> {
+       pub fn new(network: Network, feeest: Arc<FeeEstimator>, monitor: Arc<ManyChannelMonitor>, tx_broadcaster: Arc<BroadcasterInterface>, logger: Arc<Logger>,keys_manager: Arc<KeysInterface<ChanKeySigner = ChanSigner>>, config: UserConfig, current_blockchain_height: usize) -> Result<Arc<ChannelManager<ChanSigner>>, secp256k1::Error> {
                let secp_ctx = Secp256k1::new();
 
                let res = Arc::new(ChannelManager {
@@ -623,6 +624,8 @@ impl<'a> ChannelManager<'a> {
                        }),
                        our_network_key: keys_manager.get_node_secret(),
 
+                       per_peer_state: RwLock::new(HashMap::new()),
+
                        pending_events: Mutex::new(Vec::new()),
                        total_consistency_lock: RwLock::new(()),
 
@@ -733,7 +736,7 @@ impl<'a> ChannelManager<'a> {
 
                let (mut failed_htlcs, chan_option) = {
                        let mut channel_state_lock = self.channel_state.lock().unwrap();
-                       let channel_state = channel_state_lock.borrow_parts();
+                       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()?;
@@ -790,7 +793,7 @@ impl<'a> ChannelManager<'a> {
 
                let mut chan = {
                        let mut channel_state_lock = self.channel_state.lock().unwrap();
-                       let channel_state = channel_state_lock.borrow_parts();
+                       let channel_state = &mut *channel_state_lock;
                        if let Some(chan) = channel_state.by_id.remove(channel_id) {
                                if let Some(short_id) = chan.get_short_channel_id() {
                                        channel_state.short_to_id.remove(&short_id);
@@ -818,8 +821,7 @@ impl<'a> ChannelManager<'a> {
                }
        }
 
-       const ZERO:[u8; 65] = [0; 65];
-       fn decode_update_add_htlc_onion(&self, msg: &msgs::UpdateAddHTLC) -> (PendingHTLCStatus, MutexGuard<ChannelHolder>) {
+       fn decode_update_add_htlc_onion(&self, msg: &msgs::UpdateAddHTLC) -> (PendingHTLCStatus, MutexGuard<ChannelHolder<ChanSigner>>) {
                macro_rules! return_malformed_err {
                        ($msg: expr, $err_code: expr) => {
                                {
@@ -941,7 +943,7 @@ impl<'a> ChannelManager<'a> {
                        } else {
                                let mut new_packet_data = [0; 20*65];
                                chacha.process(&msg.onion_routing_packet.hop_data[65..], &mut new_packet_data[0..19*65]);
-                               chacha.process(&ChannelManager::ZERO[..], &mut new_packet_data[19*65..]);
+                               chacha.process(&SIXTY_FIVE_ZEROS[..], &mut new_packet_data[19*65..]);
 
                                let mut new_pubkey = msg.onion_routing_packet.public_key.unwrap();
 
@@ -1038,7 +1040,7 @@ impl<'a> ChannelManager<'a> {
 
        /// only fails if the channel does not yet have an assigned short_id
        /// May be called with channel_state already locked!
-       fn get_channel_update(&self, chan: &Channel) -> Result<msgs::ChannelUpdate, LightningError> {
+       fn get_channel_update(&self, chan: &Channel<ChanSigner>) -> Result<msgs::ChannelUpdate, LightningError> {
                let short_channel_id = match chan.get_short_channel_id() {
                        None => return Err(LightningError{err: "Channel not yet established", action: msgs::ErrorAction::IgnoreError}),
                        Some(id) => id,
@@ -1115,15 +1117,15 @@ impl<'a> ChannelManager<'a> {
 
                let _ = self.total_consistency_lock.read().unwrap();
 
+               let mut channel_lock = self.channel_state.lock().unwrap();
                let err: Result<(), _> = loop {
-                       let mut channel_lock = self.channel_state.lock().unwrap();
 
                        let id = match channel_lock.short_to_id.get(&route.hops.first().unwrap().short_channel_id) {
                                None => return Err(APIError::ChannelUnavailable{err: "No channel available with first hop!"}),
                                Some(id) => id.clone(),
                        };
 
-                       let channel_state = channel_lock.borrow_parts();
+                       let channel_state = &mut *channel_lock;
                        if let hash_map::Entry::Occupied(mut chan) = channel_state.by_id.entry(id) {
                                match {
                                        if chan.get().get_their_node_id() != route.hops.first().unwrap().pubkey {
@@ -1166,20 +1168,9 @@ impl<'a> ChannelManager<'a> {
                        return Ok(());
                };
 
-               match handle_error!(self, err) {
+               match handle_error!(self, err, route.hops.first().unwrap().pubkey, channel_lock) {
                        Ok(_) => unreachable!(),
-                       Err(e) => {
-                               if let msgs::ErrorAction::IgnoreError = e.action {
-                               } else {
-                                       log_error!(self, "Got bad keys: {}!", e.err);
-                                       let mut channel_state = self.channel_state.lock().unwrap();
-                                       channel_state.pending_msg_events.push(events::MessageSendEvent::HandleError {
-                                               node_id: route.hops.first().unwrap().pubkey,
-                                               action: e.action,
-                                       });
-                               }
-                               Err(APIError::ChannelUnavailable { err: e.err })
-                       },
+                       Err(e) => { Err(APIError::ChannelUnavailable { err: e.err }) }
                }
        }
 
@@ -1196,32 +1187,22 @@ impl<'a> ChannelManager<'a> {
                let _ = self.total_consistency_lock.read().unwrap();
 
                let (mut chan, msg, chan_monitor) = {
-                       let (res, chan) = {
-                               let mut channel_state = self.channel_state.lock().unwrap();
-                               match channel_state.by_id.remove(temporary_channel_id) {
-                                       Some(mut chan) => {
-                                               (chan.get_outbound_funding_created(funding_txo)
-                                                       .map_err(|e| if let ChannelError::Close(msg) = e {
-                                                               MsgHandleErrInternal::from_finish_shutdown(msg, chan.channel_id(), chan.force_shutdown(), None)
-                                                       } else { unreachable!(); })
-                                               , chan)
-                                       },
-                                       None => return
-                               }
+                       let mut channel_state = self.channel_state.lock().unwrap();
+                       let (res, chan) = match channel_state.by_id.remove(temporary_channel_id) {
+                               Some(mut chan) => {
+                                       (chan.get_outbound_funding_created(funding_txo)
+                                               .map_err(|e| if let ChannelError::Close(msg) = e {
+                                                       MsgHandleErrInternal::from_finish_shutdown(msg, chan.channel_id(), chan.force_shutdown(), None)
+                                               } else { unreachable!(); })
+                                       , chan)
+                               },
+                               None => return
                        };
-                       match handle_error!(self, res) {
+                       match handle_error!(self, res, chan.get_their_node_id(), channel_state) {
                                Ok(funding_msg) => {
                                        (chan, funding_msg.0, funding_msg.1)
                                },
-                               Err(e) => {
-                                       log_error!(self, "Got bad signatures: {}!", e.err);
-                                       let mut channel_state = self.channel_state.lock().unwrap();
-                                       channel_state.pending_msg_events.push(events::MessageSendEvent::HandleError {
-                                               node_id: chan.get_their_node_id(),
-                                               action: e.action,
-                                       });
-                                       return;
-                               },
+                               Err(_) => { return; }
                        }
                };
                // Because we have exclusive ownership of the channel here we can release the channel_state
@@ -1229,17 +1210,12 @@ impl<'a> ChannelManager<'a> {
                if let Err(e) = self.monitor.add_update_monitor(chan_monitor.get_funding_txo().unwrap(), chan_monitor) {
                        match e {
                                ChannelMonitorUpdateErr::PermanentFailure => {
-                                       match handle_error!(self, Err(MsgHandleErrInternal::from_finish_shutdown("ChannelMonitor storage failure", *temporary_channel_id, chan.force_shutdown(), None))) {
-                                               Err(e) => {
-                                                       log_error!(self, "Failed to store ChannelMonitor update for funding tx generation");
-                                                       let mut channel_state = self.channel_state.lock().unwrap();
-                                                       channel_state.pending_msg_events.push(events::MessageSendEvent::HandleError {
-                                                               node_id: chan.get_their_node_id(),
-                                                               action: e.action,
-                                                       });
-                                                       return;
-                                               },
-                                               Ok(()) => unreachable!(),
+                                       {
+                                               let mut channel_state = self.channel_state.lock().unwrap();
+                                               match handle_error!(self, Err(MsgHandleErrInternal::from_finish_shutdown("ChannelMonitor storage failure", *temporary_channel_id, chan.force_shutdown(), None)), chan.get_their_node_id(), channel_state) {
+                                                       Err(_) => { return; },
+                                                       Ok(()) => unreachable!(),
+                                               }
                                        }
                                },
                                ChannelMonitorUpdateErr::TemporaryFailure => {
@@ -1267,7 +1243,7 @@ impl<'a> ChannelManager<'a> {
                }
        }
 
-       fn get_announcement_sigs(&self, chan: &Channel) -> Option<msgs::AnnouncementSignatures> {
+       fn get_announcement_sigs(&self, chan: &Channel<ChanSigner>) -> Option<msgs::AnnouncementSignatures> {
                if !chan.should_announce() { return None }
 
                let (announcement, our_bitcoin_sig) = match chan.get_channel_announcement(self.get_our_node_id(), self.genesis_hash.clone()) {
@@ -1297,7 +1273,7 @@ impl<'a> ChannelManager<'a> {
                let mut handle_errors = Vec::new();
                {
                        let mut channel_state_lock = self.channel_state.lock().unwrap();
-                       let channel_state = channel_state_lock.borrow_parts();
+                       let channel_state = &mut *channel_state_lock;
 
                        for (short_chan_id, mut pending_forwards) in channel_state.forward_htlcs.drain() {
                                if short_chan_id != 0 {
@@ -1416,22 +1392,9 @@ impl<'a> ChannelManager<'a> {
                                                                                },
                                                                                ChannelError::CloseDelayBroadcast { .. } => { panic!("Wait is only generated on receipt of channel_reestablish, which is handled by try_chan_entry, we don't bother to support it here"); }
                                                                        };
-                                                                       match handle_error!(self, err) {
+                                                                       match handle_error!(self, err, their_node_id, channel_state) {
                                                                                Ok(_) => unreachable!(),
-                                                                               Err(e) => {
-                                                                                       match e.action {
-                                                                                               msgs::ErrorAction::IgnoreError => {},
-                                                                                               _ => {
-                                                                                                       log_error!(self, "Got bad keys: {}!", e.err);
-                                                                                                       let mut channel_state = self.channel_state.lock().unwrap();
-                                                                                                       channel_state.pending_msg_events.push(events::MessageSendEvent::HandleError {
-                                                                                                               node_id: their_node_id,
-                                                                                                               action: e.action,
-                                                                                                       });
-                                                                                               },
-                                                                                       }
-                                                                                       continue;
-                                                                               },
+                                                                               Err(_) => { continue; },
                                                                        }
                                                                }
                                                        };
@@ -1488,19 +1451,10 @@ impl<'a> ChannelManager<'a> {
                        };
                }
 
-               for (their_node_id, err) in handle_errors.drain(..) {
-                       match handle_error!(self, err) {
-                               Ok(_) => {},
-                               Err(e) => {
-                                       if let msgs::ErrorAction::IgnoreError = e.action {
-                                       } else {
-                                               let mut channel_state = self.channel_state.lock().unwrap();
-                                               channel_state.pending_msg_events.push(events::MessageSendEvent::HandleError {
-                                                       node_id: their_node_id,
-                                                       action: e.action,
-                                               });
-                                       }
-                               },
+               if handle_errors.len() > 0 {
+                       let mut channel_state_lock = self.channel_state.lock().unwrap();
+                       for (their_node_id, err) in handle_errors.drain(..) {
+                               let _ = handle_error!(self, err, their_node_id, channel_state_lock);
                        }
                }
 
@@ -1517,8 +1471,8 @@ impl<'a> ChannelManager<'a> {
        pub fn timer_chan_freshness_every_min(&self) {
                let _ = self.total_consistency_lock.read().unwrap();
                let mut channel_state_lock = self.channel_state.lock().unwrap();
-               let channel_state = channel_state_lock.borrow_parts();
-               for (_, chan) in channel_state.by_id {
+               let channel_state = &mut *channel_state_lock;
+               for (_, chan) in channel_state.by_id.iter_mut() {
                        if chan.is_disabled_staged() && !chan.is_live() {
                                if let Ok(update) = self.get_channel_update(&chan) {
                                        channel_state.pending_msg_events.push(events::MessageSendEvent::BroadcastChannelUpdate {
@@ -1561,7 +1515,7 @@ impl<'a> ChannelManager<'a> {
        /// to fail and take the channel_state lock for each iteration (as we take ownership and may
        /// drop it). In other words, no assumptions are made that entries in claimable_htlcs point to
        /// still-available channels.
-       fn fail_htlc_backwards_internal(&self, mut channel_state_lock: MutexGuard<ChannelHolder>, source: HTLCSource, payment_hash: &PaymentHash, onion_error: HTLCFailReason) {
+       fn fail_htlc_backwards_internal(&self, mut channel_state_lock: MutexGuard<ChannelHolder<ChanSigner>>, source: HTLCSource, payment_hash: &PaymentHash, onion_error: HTLCFailReason) {
                //TODO: There is a timing attack here where if a node fails an HTLC back to us they can
                //identify whether we sent it or not based on the (I presume) very different runtime
                //between the branches here. We should make this async and move it into the forward HTLCs
@@ -1689,7 +1643,7 @@ impl<'a> ChannelManager<'a> {
                        true
                } else { false }
        }
-       fn claim_funds_internal(&self, mut channel_state_lock: MutexGuard<ChannelHolder>, source: HTLCSource, payment_preimage: PaymentPreimage) {
+       fn claim_funds_internal(&self, mut channel_state_lock: MutexGuard<ChannelHolder<ChanSigner>>, source: HTLCSource, payment_preimage: PaymentPreimage) {
                let (their_node_id, err) = loop {
                        match source {
                                HTLCSource::OutboundRoute { .. } => {
@@ -1701,7 +1655,7 @@ impl<'a> ChannelManager<'a> {
                                },
                                HTLCSource::PreviousHopData(HTLCPreviousHopData { short_channel_id, htlc_id, .. }) => {
                                        //TODO: Delay the claimed_funds relaying just like we do outbound relay!
-                                       let channel_state = channel_state_lock.borrow_parts();
+                                       let channel_state = &mut *channel_state_lock;
 
                                        let chan_id = match channel_state.short_to_id.get(&short_channel_id) {
                                                Some(chan_id) => chan_id.clone(),
@@ -1753,19 +1707,7 @@ impl<'a> ChannelManager<'a> {
                        return;
                };
 
-               match handle_error!(self, err) {
-                       Ok(_) => {},
-                       Err(e) => {
-                               if let msgs::ErrorAction::IgnoreError = e.action {
-                               } else {
-                                       let mut channel_state = self.channel_state.lock().unwrap();
-                                       channel_state.pending_msg_events.push(events::MessageSendEvent::HandleError {
-                                               node_id: their_node_id,
-                                               action: e.action,
-                                       });
-                               }
-                       },
-               }
+               let _ = handle_error!(self, err, their_node_id, channel_state_lock);
        }
 
        /// Gets the node_id held by this ChannelManager
@@ -1785,12 +1727,12 @@ impl<'a> ChannelManager<'a> {
 
                {
                        let mut channel_lock = self.channel_state.lock().unwrap();
-                       let channel_state = channel_lock.borrow_parts();
-                       let short_to_id = channel_state.short_to_id;
-                       let pending_msg_events = channel_state.pending_msg_events;
+                       let channel_state = &mut *channel_lock;
+                       let short_to_id = &mut channel_state.short_to_id;
+                       let pending_msg_events = &mut channel_state.pending_msg_events;
                        channel_state.by_id.retain(|_, channel| {
                                if channel.is_awaiting_monitor_update() {
-                                       let chan_monitor = channel.channel_monitor();
+                                       let chan_monitor = channel.channel_monitor().clone();
                                        if let Err(e) = self.monitor.add_update_monitor(chan_monitor.get_funding_txo().unwrap(), chan_monitor) {
                                                match e {
                                                        ChannelMonitorUpdateErr::PermanentFailure => {
@@ -1884,15 +1826,15 @@ impl<'a> ChannelManager<'a> {
                }
        }
 
-       fn internal_open_channel(&self, their_node_id: &PublicKey, their_local_features: LocalFeatures, msg: &msgs::OpenChannel) -> Result<(), MsgHandleErrInternal> {
+       fn internal_open_channel(&self, their_node_id: &PublicKey, their_features: InitFeatures, msg: &msgs::OpenChannel) -> Result<(), MsgHandleErrInternal> {
                if msg.chain_hash != self.genesis_hash {
                        return Err(MsgHandleErrInternal::send_err_msg_no_close("Unknown genesis block hash", msg.temporary_channel_id.clone()));
                }
 
-               let channel = Channel::new_from_req(&*self.fee_estimator, &self.keys_manager, their_node_id.clone(), their_local_features, msg, 0, Arc::clone(&self.logger), &self.default_configuration)
+               let channel = Channel::new_from_req(&*self.fee_estimator, &self.keys_manager, their_node_id.clone(), their_features, msg, 0, Arc::clone(&self.logger), &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 = channel_state_lock.borrow_parts();
+               let channel_state = &mut *channel_state_lock;
                match channel_state.by_id.entry(channel.channel_id()) {
                        hash_map::Entry::Occupied(_) => return Err(MsgHandleErrInternal::send_err_msg_no_close("temporary_channel_id collision!", msg.temporary_channel_id.clone())),
                        hash_map::Entry::Vacant(entry) => {
@@ -1906,20 +1848,18 @@ impl<'a> ChannelManager<'a> {
                Ok(())
        }
 
-       fn internal_accept_channel(&self, their_node_id: &PublicKey, their_local_features: LocalFeatures, msg: &msgs::AcceptChannel) -> Result<(), MsgHandleErrInternal> {
+       fn internal_accept_channel(&self, their_node_id: &PublicKey, their_features: InitFeatures, msg: &msgs::AcceptChannel) -> Result<(), MsgHandleErrInternal> {
                let (value, output_script, user_id) = {
                        let mut channel_lock = self.channel_state.lock().unwrap();
-                       let channel_state = channel_lock.borrow_parts();
+                       let channel_state = &mut *channel_lock;
                        match channel_state.by_id.entry(msg.temporary_channel_id) {
                                hash_map::Entry::Occupied(mut chan) => {
                                        if chan.get().get_their_node_id() != *their_node_id {
-                                               //TODO: see issue #153, need a consistent behavior on obnoxious behavior from random node
                                                return Err(MsgHandleErrInternal::send_err_msg_no_close("Got a message for a channel from the wrong node!", msg.temporary_channel_id));
                                        }
-                                       try_chan_entry!(self, chan.get_mut().accept_channel(&msg, &self.default_configuration, their_local_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())
                                },
-                               //TODO: same as above
                                hash_map::Entry::Vacant(_) => return Err(MsgHandleErrInternal::send_err_msg_no_close("Failed to find corresponding channel", msg.temporary_channel_id))
                        }
                };
@@ -1936,11 +1876,10 @@ impl<'a> ChannelManager<'a> {
        fn internal_funding_created(&self, their_node_id: &PublicKey, msg: &msgs::FundingCreated) -> Result<(), MsgHandleErrInternal> {
                let ((funding_msg, monitor_update), mut chan) = {
                        let mut channel_lock = self.channel_state.lock().unwrap();
-                       let channel_state = channel_lock.borrow_parts();
+                       let channel_state = &mut *channel_lock;
                        match channel_state.by_id.entry(msg.temporary_channel_id.clone()) {
                                hash_map::Entry::Occupied(mut chan) => {
                                        if chan.get().get_their_node_id() != *their_node_id {
-                                               //TODO: here and below MsgHandleErrInternal, #153 case
                                                return Err(MsgHandleErrInternal::send_err_msg_no_close("Got a message for a channel from the wrong node!", msg.temporary_channel_id));
                                        }
                                        (try_chan_entry!(self, chan.get_mut().funding_created(msg), channel_state, chan), chan.remove())
@@ -1969,7 +1908,7 @@ impl<'a> ChannelManager<'a> {
                        }
                }
                let mut channel_state_lock = self.channel_state.lock().unwrap();
-               let channel_state = channel_state_lock.borrow_parts();
+               let channel_state = &mut *channel_state_lock;
                match channel_state.by_id.entry(funding_msg.channel_id) {
                        hash_map::Entry::Occupied(_) => {
                                return Err(MsgHandleErrInternal::send_err_msg_no_close("Already had channel with the new channel_id", funding_msg.channel_id))
@@ -1988,11 +1927,10 @@ impl<'a> ChannelManager<'a> {
        fn internal_funding_signed(&self, their_node_id: &PublicKey, msg: &msgs::FundingSigned) -> Result<(), MsgHandleErrInternal> {
                let (funding_txo, user_id) = {
                        let mut channel_lock = self.channel_state.lock().unwrap();
-                       let channel_state = channel_lock.borrow_parts();
+                       let channel_state = &mut *channel_lock;
                        match channel_state.by_id.entry(msg.channel_id) {
                                hash_map::Entry::Occupied(mut chan) => {
                                        if chan.get().get_their_node_id() != *their_node_id {
-                                               //TODO: here and below MsgHandleErrInternal, #153 case
                                                return Err(MsgHandleErrInternal::send_err_msg_no_close("Got a message for a channel from the wrong node!", msg.channel_id));
                                        }
                                        let chan_monitor = try_chan_entry!(self, chan.get_mut().funding_signed(&msg), channel_state, chan);
@@ -2014,11 +1952,10 @@ impl<'a> ChannelManager<'a> {
 
        fn internal_funding_locked(&self, their_node_id: &PublicKey, msg: &msgs::FundingLocked) -> Result<(), MsgHandleErrInternal> {
                let mut channel_state_lock = self.channel_state.lock().unwrap();
-               let channel_state = channel_state_lock.borrow_parts();
+               let channel_state = &mut *channel_state_lock;
                match channel_state.by_id.entry(msg.channel_id) {
                        hash_map::Entry::Occupied(mut chan) => {
                                if chan.get().get_their_node_id() != *their_node_id {
-                                       //TODO: here and below MsgHandleErrInternal, #153 case
                                        return Err(MsgHandleErrInternal::send_err_msg_no_close("Got a message for a channel from the wrong node!", msg.channel_id));
                                }
                                try_chan_entry!(self, chan.get_mut().funding_locked(&msg), channel_state, chan);
@@ -2046,12 +1983,11 @@ impl<'a> ChannelManager<'a> {
        fn internal_shutdown(&self, their_node_id: &PublicKey, msg: &msgs::Shutdown) -> Result<(), MsgHandleErrInternal> {
                let (mut dropped_htlcs, chan_option) = {
                        let mut channel_state_lock = self.channel_state.lock().unwrap();
-                       let channel_state = channel_state_lock.borrow_parts();
+                       let channel_state = &mut *channel_state_lock;
 
                        match channel_state.by_id.entry(msg.channel_id.clone()) {
                                hash_map::Entry::Occupied(mut chan_entry) => {
                                        if chan_entry.get().get_their_node_id() != *their_node_id {
-                                               //TODO: here and below MsgHandleErrInternal, #153 case
                                                return Err(MsgHandleErrInternal::send_err_msg_no_close("Got a message for a channel from the wrong node!", msg.channel_id));
                                        }
                                        let (shutdown, closing_signed, dropped_htlcs) = try_chan_entry!(self, chan_entry.get_mut().shutdown(&*self.fee_estimator, &msg), channel_state, chan_entry);
@@ -2094,11 +2030,10 @@ impl<'a> ChannelManager<'a> {
        fn internal_closing_signed(&self, their_node_id: &PublicKey, msg: &msgs::ClosingSigned) -> Result<(), MsgHandleErrInternal> {
                let (tx, chan_option) = {
                        let mut channel_state_lock = self.channel_state.lock().unwrap();
-                       let channel_state = channel_state_lock.borrow_parts();
+                       let channel_state = &mut *channel_state_lock;
                        match channel_state.by_id.entry(msg.channel_id.clone()) {
                                hash_map::Entry::Occupied(mut chan_entry) => {
                                        if chan_entry.get().get_their_node_id() != *their_node_id {
-                                               //TODO: here and below MsgHandleErrInternal, #153 case
                                                return Err(MsgHandleErrInternal::send_err_msg_no_close("Got a message for a channel from the wrong node!", msg.channel_id));
                                        }
                                        let (closing_signed, tx) = try_chan_entry!(self, chan_entry.get_mut().closing_signed(&*self.fee_estimator, &msg), channel_state, chan_entry);
@@ -2149,12 +2084,11 @@ impl<'a> ChannelManager<'a> {
                //but we should prevent it anyway.
 
                let (mut pending_forward_info, mut channel_state_lock) = self.decode_update_add_htlc_onion(msg);
-               let channel_state = channel_state_lock.borrow_parts();
+               let channel_state = &mut *channel_state_lock;
 
                match channel_state.by_id.entry(msg.channel_id) {
                        hash_map::Entry::Occupied(mut chan) => {
                                if chan.get().get_their_node_id() != *their_node_id {
-                                       //TODO: here MsgHandleErrInternal, #153 case
                                        return Err(MsgHandleErrInternal::send_err_msg_no_close("Got a message for a channel from the wrong node!", msg.channel_id));
                                }
                                if !chan.get().is_usable() {
@@ -2199,11 +2133,10 @@ impl<'a> ChannelManager<'a> {
        fn internal_update_fulfill_htlc(&self, their_node_id: &PublicKey, msg: &msgs::UpdateFulfillHTLC) -> Result<(), MsgHandleErrInternal> {
                let mut channel_lock = self.channel_state.lock().unwrap();
                let htlc_source = {
-                       let channel_state = channel_lock.borrow_parts();
+                       let channel_state = &mut *channel_lock;
                        match channel_state.by_id.entry(msg.channel_id) {
                                hash_map::Entry::Occupied(mut chan) => {
                                        if chan.get().get_their_node_id() != *their_node_id {
-                                               //TODO: here and below MsgHandleErrInternal, #153 case
                                                return Err(MsgHandleErrInternal::send_err_msg_no_close("Got a message for a channel from the wrong node!", msg.channel_id));
                                        }
                                        try_chan_entry!(self, chan.get_mut().update_fulfill_htlc(&msg), channel_state, chan)
@@ -2217,11 +2150,10 @@ impl<'a> ChannelManager<'a> {
 
        fn internal_update_fail_htlc(&self, their_node_id: &PublicKey, msg: &msgs::UpdateFailHTLC) -> Result<(), MsgHandleErrInternal> {
                let mut channel_lock = self.channel_state.lock().unwrap();
-               let channel_state = channel_lock.borrow_parts();
+               let channel_state = &mut *channel_lock;
                match channel_state.by_id.entry(msg.channel_id) {
                        hash_map::Entry::Occupied(mut chan) => {
                                if chan.get().get_their_node_id() != *their_node_id {
-                                       //TODO: here and below MsgHandleErrInternal, #153 case
                                        return Err(MsgHandleErrInternal::send_err_msg_no_close("Got a message for a channel from the wrong node!", msg.channel_id));
                                }
                                try_chan_entry!(self, chan.get_mut().update_fail_htlc(&msg, HTLCFailReason::LightningError { err: msg.reason.clone() }), channel_state, chan);
@@ -2233,11 +2165,10 @@ impl<'a> ChannelManager<'a> {
 
        fn internal_update_fail_malformed_htlc(&self, their_node_id: &PublicKey, msg: &msgs::UpdateFailMalformedHTLC) -> Result<(), MsgHandleErrInternal> {
                let mut channel_lock = self.channel_state.lock().unwrap();
-               let channel_state = channel_lock.borrow_parts();
+               let channel_state = &mut *channel_lock;
                match channel_state.by_id.entry(msg.channel_id) {
                        hash_map::Entry::Occupied(mut chan) => {
                                if chan.get().get_their_node_id() != *their_node_id {
-                                       //TODO: here and below MsgHandleErrInternal, #153 case
                                        return Err(MsgHandleErrInternal::send_err_msg_no_close("Got a message for a channel from the wrong node!", msg.channel_id));
                                }
                                if (msg.failure_code & 0x8000) == 0 {
@@ -2252,11 +2183,10 @@ impl<'a> ChannelManager<'a> {
 
        fn internal_commitment_signed(&self, their_node_id: &PublicKey, msg: &msgs::CommitmentSigned) -> Result<(), MsgHandleErrInternal> {
                let mut channel_state_lock = self.channel_state.lock().unwrap();
-               let channel_state = channel_state_lock.borrow_parts();
+               let channel_state = &mut *channel_state_lock;
                match channel_state.by_id.entry(msg.channel_id) {
                        hash_map::Entry::Occupied(mut chan) => {
                                if chan.get().get_their_node_id() != *their_node_id {
-                                       //TODO: here and below MsgHandleErrInternal, #153 case
                                        return Err(MsgHandleErrInternal::send_err_msg_no_close("Got a message for a channel from the wrong node!", msg.channel_id));
                                }
                                let (revoke_and_ack, commitment_signed, closing_signed, chan_monitor) =
@@ -2329,11 +2259,10 @@ impl<'a> ChannelManager<'a> {
        fn internal_revoke_and_ack(&self, their_node_id: &PublicKey, msg: &msgs::RevokeAndACK) -> Result<(), MsgHandleErrInternal> {
                let (pending_forwards, mut pending_failures, short_channel_id) = {
                        let mut channel_state_lock = self.channel_state.lock().unwrap();
-                       let channel_state = channel_state_lock.borrow_parts();
+                       let channel_state = &mut *channel_state_lock;
                        match channel_state.by_id.entry(msg.channel_id) {
                                hash_map::Entry::Occupied(mut chan) => {
                                        if chan.get().get_their_node_id() != *their_node_id {
-                                               //TODO: here and below MsgHandleErrInternal, #153 case
                                                return Err(MsgHandleErrInternal::send_err_msg_no_close("Got a message for a channel from the wrong node!", msg.channel_id));
                                        }
                                        let was_frozen_for_monitor = chan.get().is_awaiting_monitor_update();
@@ -2374,11 +2303,10 @@ impl<'a> ChannelManager<'a> {
 
        fn internal_update_fee(&self, their_node_id: &PublicKey, msg: &msgs::UpdateFee) -> Result<(), MsgHandleErrInternal> {
                let mut channel_lock = self.channel_state.lock().unwrap();
-               let channel_state = channel_lock.borrow_parts();
+               let channel_state = &mut *channel_lock;
                match channel_state.by_id.entry(msg.channel_id) {
                        hash_map::Entry::Occupied(mut chan) => {
                                if chan.get().get_their_node_id() != *their_node_id {
-                                       //TODO: here and below MsgHandleErrInternal, #153 case
                                        return Err(MsgHandleErrInternal::send_err_msg_no_close("Got a message for a channel from the wrong node!", msg.channel_id));
                                }
                                try_chan_entry!(self, chan.get_mut().update_fee(&*self.fee_estimator, &msg), channel_state, chan);
@@ -2390,7 +2318,7 @@ impl<'a> ChannelManager<'a> {
 
        fn internal_announcement_signatures(&self, their_node_id: &PublicKey, msg: &msgs::AnnouncementSignatures) -> Result<(), MsgHandleErrInternal> {
                let mut channel_state_lock = self.channel_state.lock().unwrap();
-               let channel_state = channel_state_lock.borrow_parts();
+               let channel_state = &mut *channel_state_lock;
 
                match channel_state.by_id.entry(msg.channel_id) {
                        hash_map::Entry::Occupied(mut chan) => {
@@ -2432,7 +2360,7 @@ impl<'a> ChannelManager<'a> {
 
        fn internal_channel_reestablish(&self, their_node_id: &PublicKey, msg: &msgs::ChannelReestablish) -> Result<(), MsgHandleErrInternal> {
                let mut channel_state_lock = self.channel_state.lock().unwrap();
-               let channel_state = channel_state_lock.borrow_parts();
+               let channel_state = &mut *channel_state_lock;
 
                match channel_state.by_id.entry(msg.channel_id) {
                        hash_map::Entry::Occupied(mut chan) => {
@@ -2507,10 +2435,10 @@ impl<'a> ChannelManager<'a> {
        #[doc(hidden)]
        pub fn update_fee(&self, channel_id: [u8;32], feerate_per_kw: u64) -> Result<(), APIError> {
                let _ = self.total_consistency_lock.read().unwrap();
+               let mut channel_state_lock = self.channel_state.lock().unwrap();
                let their_node_id;
                let err: Result<(), _> = loop {
-                       let mut channel_state_lock = self.channel_state.lock().unwrap();
-                       let channel_state = channel_state_lock.borrow_parts();
+                       let channel_state = &mut *channel_state_lock;
 
                        match channel_state.by_id.entry(channel_id) {
                                hash_map::Entry::Vacant(_) => return Err(APIError::APIMisuseError{err: "Failed to find corresponding channel"}),
@@ -2548,25 +2476,14 @@ impl<'a> ChannelManager<'a> {
                        return Ok(())
                };
 
-               match handle_error!(self, err) {
+               match handle_error!(self, err, their_node_id, channel_state_lock) {
                        Ok(_) => unreachable!(),
-                       Err(e) => {
-                               if let msgs::ErrorAction::IgnoreError = e.action {
-                               } else {
-                                       log_error!(self, "Got bad keys: {}!", e.err);
-                                       let mut channel_state = self.channel_state.lock().unwrap();
-                                       channel_state.pending_msg_events.push(events::MessageSendEvent::HandleError {
-                                               node_id: their_node_id,
-                                               action: e.action,
-                                       });
-                               }
-                               Err(APIError::APIMisuseError { err: e.err })
-                       },
+                       Err(e) => { Err(APIError::APIMisuseError { err: e.err })}
                }
        }
 }
 
-impl<'a> events::MessageSendEventsProvider for ChannelManager<'a> {
+impl<ChanSigner: ChannelKeys> events::MessageSendEventsProvider for ChannelManager<ChanSigner> {
        fn get_and_clear_pending_msg_events(&self) -> Vec<events::MessageSendEvent> {
                // TODO: Event release to users and serialization is currently race-y: it's very easy for a
                // user to serialize a ChannelManager with pending events in it and lose those events on
@@ -2591,7 +2508,7 @@ impl<'a> events::MessageSendEventsProvider for ChannelManager<'a> {
        }
 }
 
-impl<'a> events::EventsProvider for ChannelManager<'a> {
+impl<ChanSigner: ChannelKeys> events::EventsProvider for ChannelManager<ChanSigner> {
        fn get_and_clear_pending_events(&self) -> Vec<events::Event> {
                // TODO: Event release to users and serialization is currently race-y: it's very easy for a
                // user to serialize a ChannelManager with pending events in it and lose those events on
@@ -2616,7 +2533,7 @@ impl<'a> events::EventsProvider for ChannelManager<'a> {
        }
 }
 
-impl<'a> ChainListener for ChannelManager<'a> {
+impl<ChanSigner: ChannelKeys> ChainListener for ChannelManager<ChanSigner> {
        fn block_connected(&self, header: &BlockHeader, height: u32, txn_matched: &[&Transaction], indexes_of_txn_matched: &[u32]) {
                let header_hash = header.bitcoin_hash();
                log_trace!(self, "Block {} at height {} connected with {} txn matched", header_hash, height, txn_matched.len());
@@ -2624,9 +2541,9 @@ impl<'a> ChainListener for ChannelManager<'a> {
                let mut failed_channels = Vec::new();
                {
                        let mut channel_lock = self.channel_state.lock().unwrap();
-                       let channel_state = channel_lock.borrow_parts();
-                       let short_to_id = channel_state.short_to_id;
-                       let pending_msg_events = channel_state.pending_msg_events;
+                       let channel_state = &mut *channel_lock;
+                       let short_to_id = &mut channel_state.short_to_id;
+                       let pending_msg_events = &mut channel_state.pending_msg_events;
                        channel_state.by_id.retain(|_, channel| {
                                let chan_res = channel.block_connected(header, height, txn_matched, indexes_of_txn_matched);
                                if let Ok(Some(funding_locked)) = chan_res {
@@ -2702,9 +2619,9 @@ impl<'a> ChainListener for ChannelManager<'a> {
                let mut failed_channels = Vec::new();
                {
                        let mut channel_lock = self.channel_state.lock().unwrap();
-                       let channel_state = channel_lock.borrow_parts();
-                       let short_to_id = channel_state.short_to_id;
-                       let pending_msg_events = channel_state.pending_msg_events;
+                       let channel_state = &mut *channel_lock;
+                       let short_to_id = &mut channel_state.short_to_id;
+                       let pending_msg_events = &mut channel_state.pending_msg_events;
                        channel_state.by_id.retain(|_,  v| {
                                if v.block_disconnected(header) {
                                        if let Some(short_id) = v.get_short_channel_id() {
@@ -2730,97 +2647,161 @@ impl<'a> ChainListener for ChannelManager<'a> {
        }
 }
 
-impl<'a> ChannelMessageHandler for ChannelManager<'a> {
-       //TODO: Handle errors and close channel (or so)
-       fn handle_open_channel(&self, their_node_id: &PublicKey, their_local_features: LocalFeatures, msg: &msgs::OpenChannel) -> Result<(), LightningError> {
+impl<ChanSigner: ChannelKeys> ChannelMessageHandler for ChannelManager<ChanSigner> {
+       fn handle_open_channel(&self, their_node_id: &PublicKey, their_features: InitFeatures, msg: &msgs::OpenChannel) {
                let _ = self.total_consistency_lock.read().unwrap();
-               handle_error!(self, self.internal_open_channel(their_node_id, their_local_features, msg))
+               let res = self.internal_open_channel(their_node_id, their_features, msg);
+               if res.is_err() {
+                       let mut channel_state_lock = self.channel_state.lock().unwrap();
+                       let _ = handle_error!(self, res, *their_node_id, channel_state_lock);
+               }
        }
 
-       fn handle_accept_channel(&self, their_node_id: &PublicKey, their_local_features: LocalFeatures, msg: &msgs::AcceptChannel) -> Result<(), LightningError> {
+       fn handle_accept_channel(&self, their_node_id: &PublicKey, their_features: InitFeatures, msg: &msgs::AcceptChannel) {
                let _ = self.total_consistency_lock.read().unwrap();
-               handle_error!(self, self.internal_accept_channel(their_node_id, their_local_features, msg))
+               let res = self.internal_accept_channel(their_node_id, their_features, msg);
+               if res.is_err() {
+                       let mut channel_state_lock = self.channel_state.lock().unwrap();
+                       let _ = handle_error!(self, res, *their_node_id, channel_state_lock);
+               }
        }
 
-       fn handle_funding_created(&self, their_node_id: &PublicKey, msg: &msgs::FundingCreated) -> Result<(), LightningError> {
+       fn handle_funding_created(&self, their_node_id: &PublicKey, msg: &msgs::FundingCreated) {
                let _ = self.total_consistency_lock.read().unwrap();
-               handle_error!(self, self.internal_funding_created(their_node_id, msg))
+               let res = self.internal_funding_created(their_node_id, msg);
+               if res.is_err() {
+                       let mut channel_state_lock = self.channel_state.lock().unwrap();
+                       let _ = handle_error!(self, res, *their_node_id, channel_state_lock);
+               }
        }
 
-       fn handle_funding_signed(&self, their_node_id: &PublicKey, msg: &msgs::FundingSigned) -> Result<(), LightningError> {
+       fn handle_funding_signed(&self, their_node_id: &PublicKey, msg: &msgs::FundingSigned) {
                let _ = self.total_consistency_lock.read().unwrap();
-               handle_error!(self, self.internal_funding_signed(their_node_id, msg))
+               let res = self.internal_funding_signed(their_node_id, msg);
+               if res.is_err() {
+                       let mut channel_state_lock = self.channel_state.lock().unwrap();
+                       let _ = handle_error!(self, res, *their_node_id, channel_state_lock);
+               }
        }
 
-       fn handle_funding_locked(&self, their_node_id: &PublicKey, msg: &msgs::FundingLocked) -> Result<(), LightningError> {
+       fn handle_funding_locked(&self, their_node_id: &PublicKey, msg: &msgs::FundingLocked) {
                let _ = self.total_consistency_lock.read().unwrap();
-               handle_error!(self, self.internal_funding_locked(their_node_id, msg))
+               let res = self.internal_funding_locked(their_node_id, msg);
+               if res.is_err() {
+                       let mut channel_state_lock = self.channel_state.lock().unwrap();
+                       let _ = handle_error!(self, res, *their_node_id, channel_state_lock);
+               }
        }
 
-       fn handle_shutdown(&self, their_node_id: &PublicKey, msg: &msgs::Shutdown) -> Result<(), LightningError> {
+       fn handle_shutdown(&self, their_node_id: &PublicKey, msg: &msgs::Shutdown) {
                let _ = self.total_consistency_lock.read().unwrap();
-               handle_error!(self, self.internal_shutdown(their_node_id, msg))
+               let res = self.internal_shutdown(their_node_id, msg);
+               if res.is_err() {
+                       let mut channel_state_lock = self.channel_state.lock().unwrap();
+                       let _ = handle_error!(self, res, *their_node_id, channel_state_lock);
+               }
        }
 
-       fn handle_closing_signed(&self, their_node_id: &PublicKey, msg: &msgs::ClosingSigned) -> Result<(), LightningError> {
+       fn handle_closing_signed(&self, their_node_id: &PublicKey, msg: &msgs::ClosingSigned) {
                let _ = self.total_consistency_lock.read().unwrap();
-               handle_error!(self, self.internal_closing_signed(their_node_id, msg))
+               let res = self.internal_closing_signed(their_node_id, msg);
+               if res.is_err() {
+                       let mut channel_state_lock = self.channel_state.lock().unwrap();
+                       let _ = handle_error!(self, res, *their_node_id, channel_state_lock);
+               }
        }
 
-       fn handle_update_add_htlc(&self, their_node_id: &PublicKey, msg: &msgs::UpdateAddHTLC) -> Result<(), LightningError> {
+       fn handle_update_add_htlc(&self, their_node_id: &PublicKey, msg: &msgs::UpdateAddHTLC) {
                let _ = self.total_consistency_lock.read().unwrap();
-               handle_error!(self, self.internal_update_add_htlc(their_node_id, msg))
+               let res = self.internal_update_add_htlc(their_node_id, msg);
+               if res.is_err() {
+                       let mut channel_state_lock = self.channel_state.lock().unwrap();
+                       let _ = handle_error!(self, res, *their_node_id, channel_state_lock);
+               }
        }
 
-       fn handle_update_fulfill_htlc(&self, their_node_id: &PublicKey, msg: &msgs::UpdateFulfillHTLC) -> Result<(), LightningError> {
+       fn handle_update_fulfill_htlc(&self, their_node_id: &PublicKey, msg: &msgs::UpdateFulfillHTLC) {
                let _ = self.total_consistency_lock.read().unwrap();
-               handle_error!(self, self.internal_update_fulfill_htlc(their_node_id, msg))
+               let res = self.internal_update_fulfill_htlc(their_node_id, msg);
+               if res.is_err() {
+                       let mut channel_state_lock = self.channel_state.lock().unwrap();
+                       let _ = handle_error!(self, res, *their_node_id, channel_state_lock);
+               }
        }
 
-       fn handle_update_fail_htlc(&self, their_node_id: &PublicKey, msg: &msgs::UpdateFailHTLC) -> Result<(), LightningError> {
+       fn handle_update_fail_htlc(&self, their_node_id: &PublicKey, msg: &msgs::UpdateFailHTLC) {
                let _ = self.total_consistency_lock.read().unwrap();
-               handle_error!(self, self.internal_update_fail_htlc(their_node_id, msg))
+               let res = self.internal_update_fail_htlc(their_node_id, msg);
+               if res.is_err() {
+                       let mut channel_state_lock = self.channel_state.lock().unwrap();
+                       let _ = handle_error!(self, res, *their_node_id, channel_state_lock);
+               }
        }
 
-       fn handle_update_fail_malformed_htlc(&self, their_node_id: &PublicKey, msg: &msgs::UpdateFailMalformedHTLC) -> Result<(), LightningError> {
+       fn handle_update_fail_malformed_htlc(&self, their_node_id: &PublicKey, msg: &msgs::UpdateFailMalformedHTLC) {
                let _ = self.total_consistency_lock.read().unwrap();
-               handle_error!(self, self.internal_update_fail_malformed_htlc(their_node_id, msg))
+               let res = self.internal_update_fail_malformed_htlc(their_node_id, msg);
+               if res.is_err() {
+                       let mut channel_state_lock = self.channel_state.lock().unwrap();
+                       let _ = handle_error!(self, res, *their_node_id, channel_state_lock);
+               }
        }
 
-       fn handle_commitment_signed(&self, their_node_id: &PublicKey, msg: &msgs::CommitmentSigned) -> Result<(), LightningError> {
+       fn handle_commitment_signed(&self, their_node_id: &PublicKey, msg: &msgs::CommitmentSigned) {
                let _ = self.total_consistency_lock.read().unwrap();
-               handle_error!(self, self.internal_commitment_signed(their_node_id, msg))
+               let res = self.internal_commitment_signed(their_node_id, msg);
+               if res.is_err() {
+                       let mut channel_state_lock = self.channel_state.lock().unwrap();
+                       let _ = handle_error!(self, res, *their_node_id, channel_state_lock);
+               }
        }
 
-       fn handle_revoke_and_ack(&self, their_node_id: &PublicKey, msg: &msgs::RevokeAndACK) -> Result<(), LightningError> {
+       fn handle_revoke_and_ack(&self, their_node_id: &PublicKey, msg: &msgs::RevokeAndACK) {
                let _ = self.total_consistency_lock.read().unwrap();
-               handle_error!(self, self.internal_revoke_and_ack(their_node_id, msg))
+               let res = self.internal_revoke_and_ack(their_node_id, msg);
+               if res.is_err() {
+                       let mut channel_state_lock = self.channel_state.lock().unwrap();
+                       let _ = handle_error!(self, res, *their_node_id, channel_state_lock);
+               }
        }
 
-       fn handle_update_fee(&self, their_node_id: &PublicKey, msg: &msgs::UpdateFee) -> Result<(), LightningError> {
+       fn handle_update_fee(&self, their_node_id: &PublicKey, msg: &msgs::UpdateFee) {
                let _ = self.total_consistency_lock.read().unwrap();
-               handle_error!(self, self.internal_update_fee(their_node_id, msg))
+               let res = self.internal_update_fee(their_node_id, msg);
+               if res.is_err() {
+                       let mut channel_state_lock = self.channel_state.lock().unwrap();
+                       let _ = handle_error!(self, res, *their_node_id, channel_state_lock);
+               }
        }
 
-       fn handle_announcement_signatures(&self, their_node_id: &PublicKey, msg: &msgs::AnnouncementSignatures) -> Result<(), LightningError> {
+       fn handle_announcement_signatures(&self, their_node_id: &PublicKey, msg: &msgs::AnnouncementSignatures) {
                let _ = self.total_consistency_lock.read().unwrap();
-               handle_error!(self, self.internal_announcement_signatures(their_node_id, msg))
+               let res = self.internal_announcement_signatures(their_node_id, msg);
+               if res.is_err() {
+                       let mut channel_state_lock = self.channel_state.lock().unwrap();
+                       let _ = handle_error!(self, res, *their_node_id, channel_state_lock);
+               }
        }
 
-       fn handle_channel_reestablish(&self, their_node_id: &PublicKey, msg: &msgs::ChannelReestablish) -> Result<(), LightningError> {
+       fn handle_channel_reestablish(&self, their_node_id: &PublicKey, msg: &msgs::ChannelReestablish) {
                let _ = self.total_consistency_lock.read().unwrap();
-               handle_error!(self, self.internal_channel_reestablish(their_node_id, msg))
+               let res = self.internal_channel_reestablish(their_node_id, msg);
+               if res.is_err() {
+                       let mut channel_state_lock = self.channel_state.lock().unwrap();
+                       let _ = handle_error!(self, res, *their_node_id, channel_state_lock);
+               }
        }
 
        fn peer_disconnected(&self, their_node_id: &PublicKey, no_connection_possible: bool) {
                let _ = self.total_consistency_lock.read().unwrap();
                let mut failed_channels = Vec::new();
                let mut failed_payments = Vec::new();
+               let mut no_channels_remain = true;
                {
                        let mut channel_state_lock = self.channel_state.lock().unwrap();
-                       let channel_state = channel_state_lock.borrow_parts();
-                       let short_to_id = channel_state.short_to_id;
-                       let pending_msg_events = channel_state.pending_msg_events;
+                       let channel_state = &mut *channel_state_lock;
+                       let short_to_id = &mut channel_state.short_to_id;
+                       let pending_msg_events = &mut channel_state.pending_msg_events;
                        if no_connection_possible {
                                log_debug!(self, "Failing all channels with {} due to no_connection_possible", log_pubkey!(their_node_id));
                                channel_state.by_id.retain(|_, chan| {
@@ -2854,6 +2835,8 @@ impl<'a> ChannelMessageHandler for ChannelManager<'a> {
                                                                short_to_id.remove(&short_id);
                                                        }
                                                        return false;
+                                               } else {
+                                                       no_channels_remain = false;
                                                }
                                        }
                                        true
@@ -2879,6 +2862,10 @@ impl<'a> ChannelMessageHandler for ChannelManager<'a> {
                                }
                        });
                }
+               if no_channels_remain {
+                       self.per_peer_state.write().unwrap().remove(their_node_id);
+               }
+
                for failure in failed_channels.drain(..) {
                        self.finish_force_close_channel(failure);
                }
@@ -2889,13 +2876,28 @@ impl<'a> ChannelMessageHandler for ChannelManager<'a> {
                }
        }
 
-       fn peer_connected(&self, their_node_id: &PublicKey) {
+       fn peer_connected(&self, their_node_id: &PublicKey, init_msg: &msgs::Init) {
                log_debug!(self, "Generating channel_reestablish events for {}", log_pubkey!(their_node_id));
 
                let _ = self.total_consistency_lock.read().unwrap();
+
+               {
+                       let mut peer_state_lock = self.per_peer_state.write().unwrap();
+                       match peer_state_lock.entry(their_node_id.clone()) {
+                               hash_map::Entry::Vacant(e) => {
+                                       e.insert(Mutex::new(PeerState {
+                                               latest_features: init_msg.features.clone(),
+                                       }));
+                               },
+                               hash_map::Entry::Occupied(e) => {
+                                       e.get().lock().unwrap().latest_features = init_msg.features.clone();
+                               },
+                       }
+               }
+
                let mut channel_state_lock = self.channel_state.lock().unwrap();
-               let channel_state = channel_state_lock.borrow_parts();
-               let pending_msg_events = channel_state.pending_msg_events;
+               let channel_state = &mut *channel_state_lock;
+               let pending_msg_events = &mut channel_state.pending_msg_events;
                channel_state.by_id.retain(|_, chan| {
                        if chan.get_their_node_id() == *their_node_id {
                                if !chan.have_received_message() {
@@ -3115,7 +3117,7 @@ impl<R: ::std::io::Read> Readable<R> for HTLCForwardInfo {
        }
 }
 
-impl<'a> Writeable for ChannelManager<'a> {
+impl<ChanSigner: ChannelKeys + Writeable> Writeable for ChannelManager<ChanSigner> {
        fn write<W: Writer>(&self, writer: &mut W) -> Result<(), ::std::io::Error> {
                let _ = self.total_consistency_lock.write().unwrap();
 
@@ -3159,6 +3161,14 @@ impl<'a> Writeable for ChannelManager<'a> {
                        }
                }
 
+               let per_peer_state = self.per_peer_state.write().unwrap();
+               (per_peer_state.len() as u64).write(writer)?;
+               for (peer_pubkey, peer_state_mutex) in per_peer_state.iter() {
+                       peer_pubkey.write(writer)?;
+                       let peer_state = peer_state_mutex.lock().unwrap();
+                       peer_state.latest_features.write(writer)?;
+               }
+
                Ok(())
        }
 }
@@ -3178,10 +3188,10 @@ impl<'a> Writeable for ChannelManager<'a> {
 /// 5) Move the ChannelMonitors into your local ManyChannelMonitor.
 /// 6) Disconnect/connect blocks on the ChannelManager.
 /// 7) Register the new ChannelManager with your ChainWatchInterface.
-pub struct ChannelManagerReadArgs<'a, 'b> {
+pub struct ChannelManagerReadArgs<'a, ChanSigner: ChannelKeys> {
        /// The keys provider which will give us relevant keys. Some keys will be loaded during
        /// deserialization.
-       pub keys_manager: Arc<KeysInterface>,
+       pub keys_manager: Arc<KeysInterface<ChanKeySigner = ChanSigner>>,
 
        /// The fee_estimator for use in the ChannelManager in the future.
        ///
@@ -3192,7 +3202,7 @@ pub struct ChannelManagerReadArgs<'a, 'b> {
        /// No calls to the ManyChannelMonitor will be made during deserialization. It is assumed that
        /// you have deserialized ChannelMonitors separately and will add them to your
        /// ManyChannelMonitor after deserializing this ChannelManager.
-       pub monitor: Arc<ManyChannelMonitor + 'b>,
+       pub monitor: Arc<ManyChannelMonitor>,
 
        /// The BroadcasterInterface which will be used in the ChannelManager in the future and may be
        /// used to broadcast the latest local commitment transactions of channels which must be
@@ -3215,11 +3225,11 @@ pub struct ChannelManagerReadArgs<'a, 'b> {
        ///
        /// In such cases the latest local transactions will be sent to the tx_broadcaster included in
        /// this struct.
-       pub channel_monitors: &'a HashMap<OutPoint, &'a ChannelMonitor>,
+       pub channel_monitors: &'a mut HashMap<OutPoint, &'a mut ChannelMonitor>,
 }
 
-impl<'a, 'b, R : ::std::io::Read> ReadableArgs<R, ChannelManagerReadArgs<'a, 'b>> for (Sha256dHash, ChannelManager<'b>) {
-       fn read(reader: &mut R, args: ChannelManagerReadArgs<'a, 'b>) -> Result<Self, DecodeError> {
+impl<'a, R : ::std::io::Read, ChanSigner: ChannelKeys + Readable<R>> ReadableArgs<R, ChannelManagerReadArgs<'a, ChanSigner>> for (Sha256dHash, ChannelManager<ChanSigner>) {
+       fn read(reader: &mut R, args: ChannelManagerReadArgs<'a, ChanSigner>) -> Result<Self, DecodeError> {
                let _ver: u8 = Readable::read(reader)?;
                let min_ver: u8 = Readable::read(reader)?;
                if min_ver > SERIALIZATION_VERSION {
@@ -3237,14 +3247,14 @@ impl<'a, 'b, R : ::std::io::Read> ReadableArgs<R, ChannelManagerReadArgs<'a, 'b>
                let mut by_id = HashMap::with_capacity(cmp::min(channel_count as usize, 128));
                let mut short_to_id = HashMap::with_capacity(cmp::min(channel_count as usize, 128));
                for _ in 0..channel_count {
-                       let mut channel: Channel = ReadableArgs::read(reader, args.logger.clone())?;
+                       let mut channel: Channel<ChanSigner> = ReadableArgs::read(reader, args.logger.clone())?;
                        if channel.last_block_connected != last_block_hash {
                                return Err(DecodeError::InvalidValue);
                        }
 
                        let funding_txo = channel.channel_monitor().get_funding_txo().ok_or(DecodeError::InvalidValue)?;
                        funding_txo_set.insert(funding_txo.clone());
-                       if let Some(monitor) = args.channel_monitors.get(&funding_txo) {
+                       if let Some(ref mut monitor) = args.channel_monitors.get_mut(&funding_txo) {
                                if channel.get_cur_local_commitment_transaction_number() != monitor.get_cur_local_commitment_number() ||
                                                channel.get_revoked_remote_commitment_transaction_number() != monitor.get_min_seen_secret() ||
                                                channel.get_cur_remote_commitment_transaction_number() != monitor.get_cur_remote_commitment_number() {
@@ -3262,7 +3272,7 @@ impl<'a, 'b, R : ::std::io::Read> ReadableArgs<R, ChannelManagerReadArgs<'a, 'b>
                        }
                }
 
-               for (ref funding_txo, ref monitor) in args.channel_monitors.iter() {
+               for (ref funding_txo, ref mut monitor) in args.channel_monitors.iter_mut() {
                        if !funding_txo_set.contains(funding_txo) {
                                closed_channels.push((monitor.get_latest_local_commitment_txn(), Vec::new()));
                        }
@@ -3292,6 +3302,16 @@ impl<'a, 'b, R : ::std::io::Read> ReadableArgs<R, ChannelManagerReadArgs<'a, 'b>
                        claimable_htlcs.insert(payment_hash, previous_hops);
                }
 
+               let peer_count: u64 = Readable::read(reader)?;
+               let mut per_peer_state = HashMap::with_capacity(cmp::min(peer_count as usize, 128));
+               for _ in 0..peer_count {
+                       let peer_pubkey = Readable::read(reader)?;
+                       let peer_state = PeerState {
+                               latest_features: Readable::read(reader)?,
+                       };
+                       per_peer_state.insert(peer_pubkey, Mutex::new(peer_state));
+               }
+
                let channel_manager = ChannelManager {
                        genesis_hash,
                        fee_estimator: args.fee_estimator,
@@ -3311,6 +3331,8 @@ impl<'a, 'b, R : ::std::io::Read> ReadableArgs<R, ChannelManagerReadArgs<'a, 'b>
                        }),
                        our_network_key: args.keys_manager.get_node_secret(),
 
+                       per_peer_state: RwLock::new(per_peer_state),
+
                        pending_events: Mutex::new(Vec::new()),
                        total_consistency_lock: RwLock::new(()),
                        keys_manager: args.keys_manager,