Rename HandleError to LightningError to stress already-processed error
[rust-lightning] / src / ln / channelmanager.rs
index 64a9f2ebb8e575c3895d77efd3ab4eaa1b82ccf0..156b31964887fc1c7ae3af85e250913cd4bab301 100644 (file)
@@ -31,11 +31,12 @@ 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::msgs;
+use ln::msgs::LocalFeatures;
 use ln::onion_utils;
-use ln::msgs::{ChannelMessageHandler, DecodeError, HandleError};
+use ln::msgs::{ChannelMessageHandler, DecodeError, LightningError};
 use chain::keysinterface::KeysInterface;
 use util::config::UserConfig;
-use util::{byte_utils, events, rng};
+use util::{byte_utils, events};
 use util::ser::{Readable, ReadableArgs, Writeable, Writer};
 use util::chacha20::ChaCha20;
 use util::logger::Logger;
@@ -46,7 +47,7 @@ use std::collections::{HashMap, hash_map, HashSet};
 use std::io::Cursor;
 use std::sync::{Arc, Mutex, MutexGuard, RwLock};
 use std::sync::atomic::{AtomicUsize, Ordering};
-use std::time::{Instant,Duration};
+use std::time::Duration;
 
 // We hold various information about HTLC relay in the HTLC objects in Channel itself:
 //
@@ -117,7 +118,7 @@ impl HTLCSource {
 
 #[derive(Clone)] // See Channel::revoke_and_ack for why, tl;dr: Rust bug
 pub(super) enum HTLCFailReason {
-       ErrorPacket {
+       LightningError {
                err: msgs::OnionErrorPacket,
        },
        Reason {
@@ -142,14 +143,14 @@ type ShutdownResult = (Vec<Transaction>, Vec<(HTLCSource, PaymentHash)>);
 /// this struct and call handle_error!() on it.
 
 struct MsgHandleErrInternal {
-       err: msgs::HandleError,
+       err: msgs::LightningError,
        shutdown_finish: Option<(ShutdownResult, Option<msgs::ChannelUpdate>)>,
 }
 impl MsgHandleErrInternal {
        #[inline]
        fn send_err_msg_no_close(err: &'static str, channel_id: [u8; 32]) -> Self {
                Self {
-                       err: HandleError {
+                       err: LightningError {
                                err,
                                action: Some(msgs::ErrorAction::SendErrorMessage {
                                        msg: msgs::ErrorMessage {
@@ -164,7 +165,7 @@ impl MsgHandleErrInternal {
        #[inline]
        fn ignore_no_close(err: &'static str) -> Self {
                Self {
-                       err: HandleError {
+                       err: LightningError {
                                err,
                                action: Some(msgs::ErrorAction::IgnoreError),
                        },
@@ -172,13 +173,13 @@ impl MsgHandleErrInternal {
                }
        }
        #[inline]
-       fn from_no_close(err: msgs::HandleError) -> Self {
+       fn from_no_close(err: msgs::LightningError) -> Self {
                Self { err, shutdown_finish: None }
        }
        #[inline]
        fn from_finish_shutdown(err: &'static str, channel_id: [u8; 32], shutdown_res: ShutdownResult, channel_update: Option<msgs::ChannelUpdate>) -> Self {
                Self {
-                       err: HandleError {
+                       err: LightningError {
                                err,
                                action: Some(msgs::ErrorAction::SendErrorMessage {
                                        msg: msgs::ErrorMessage {
@@ -194,11 +195,20 @@ impl MsgHandleErrInternal {
        fn from_chan_no_close(err: ChannelError, channel_id: [u8; 32]) -> Self {
                Self {
                        err: match err {
-                               ChannelError::Ignore(msg) => HandleError {
+                               ChannelError::Ignore(msg) => LightningError {
                                        err: msg,
                                        action: Some(msgs::ErrorAction::IgnoreError),
                                },
-                               ChannelError::Close(msg) => HandleError {
+                               ChannelError::Close(msg) => LightningError {
+                                       err: msg,
+                                       action: Some(msgs::ErrorAction::SendErrorMessage {
+                                               msg: msgs::ErrorMessage {
+                                                       channel_id,
+                                                       data: msg.to_string()
+                                               },
+                                       }),
+                               },
+                               ChannelError::CloseDelayBroadcast { msg, .. } => LightningError {
                                        err: msg,
                                        action: Some(msgs::ErrorAction::SendErrorMessage {
                                                msg: msgs::ErrorMessage {
@@ -213,11 +223,11 @@ impl MsgHandleErrInternal {
        }
 }
 
-/// We hold back HTLCs we intend to relay for a random interval in the range (this, 5*this). This
-/// provides some limited amount of privacy. Ideally this would range from somewhere like 1 second
-/// to 30 seconds, but people expect lightning to be, you know, kinda fast, sadly. We could
-/// probably increase this significantly.
-const MIN_HTLC_RELAY_HOLDING_CELL_MILLIS: u32 = 50;
+/// We hold back HTLCs we intend to relay for a random interval greater than this (see
+/// Event::PendingHTLCsForwardable for the API guidelines indicating how long should be waited).
+/// This provides some limited amount of privacy. Ideally this would range from somewhere like one
+/// second to 30 seconds, but people expect lightning to be, you know, kinda fast, sadly.
+const MIN_HTLC_RELAY_HOLDING_CELL_MILLIS: u64 = 100;
 
 pub(super) enum HTLCForwardInfo {
        AddHTLC {
@@ -247,7 +257,6 @@ pub(super) enum RAACommitmentOrder {
 pub(super) struct ChannelHolder {
        pub(super) by_id: HashMap<[u8; 32], Channel>,
        pub(super) short_to_id: HashMap<u64, [u8; 32]>,
-       pub(super) next_forward: Instant,
        /// 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
        /// guarantees are made about the existence of a channel with the short id here, nor the short
@@ -266,7 +275,6 @@ pub(super) struct ChannelHolder {
 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) next_forward: &'a mut Instant,
        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>,
@@ -276,7 +284,6 @@ impl ChannelHolder {
                MutChannelHolder {
                        by_id: &mut self.by_id,
                        short_to_id: &mut self.short_to_id,
-                       next_forward: &mut self.next_forward,
                        forward_htlcs: &mut self.forward_htlcs,
                        claimable_htlcs: &mut self.claimable_htlcs,
                        pending_msg_events: &mut self.pending_msg_events,
@@ -343,6 +350,12 @@ pub struct ChannelManager {
        logger: Arc<Logger>,
 }
 
+/// The amount of time we require our counterparty wait to claim their money (ie time between when
+/// we, or our watchtower, must check for them having broadcast a theft transaction).
+pub(crate) const BREAKDOWN_TIMEOUT: u16 = 6 * 24;
+/// The amount of time we're willing to wait to claim money back to us
+pub(crate) const MAX_LOCAL_BREAKDOWN_TIMEOUT: u16 = 6 * 24 * 7;
+
 /// The minimum number of blocks between an inbound HTLC's CLTV and the corresponding outbound
 /// HTLC's CLTV. This should always be a few blocks greater than channelmonitor::CLTV_CLAIM_BUFFER,
 /// ie the node we forwarded the payment on to should always have enough room to reliably time out
@@ -392,6 +405,20 @@ pub struct ChannelDetails {
        pub channel_value_satoshis: u64,
        /// The user_id passed in to create_channel, or 0 if the channel was inbound.
        pub user_id: u64,
+       /// The available outbound capacity for sending HTLCs to the remote peer. This does not include
+       /// any pending HTLCs which are not yet fully resolved (and, thus, who's balance is not
+       /// available for inclusion in new outbound HTLCs). This further does not include any pending
+       /// outgoing HTLCs which are awaiting some other resolution to be sent.
+       pub outbound_capacity_msat: u64,
+       /// The available inbound capacity for the remote peer to send HTLCs to us. This does not
+       /// include any pending HTLCs which are not yet fully resolved (and, thus, who's balance is not
+       /// available for inclusion in new inbound HTLCs).
+       /// Note that there are some corner cases not fully handled here, so the actual available
+       /// inbound capacity may be slightly higher than this.
+       pub inbound_capacity_msat: u64,
+       /// True if the channel is (a) confirmed and funding_locked messages have been exchanged, (b)
+       /// the peer is connected, and (c) no monitor update failure is pending resolution.
+       pub is_live: bool,
 }
 
 macro_rules! handle_error {
@@ -429,6 +456,7 @@ macro_rules! break_chan_entry {
                                }
                                break Err(MsgHandleErrInternal::from_finish_shutdown(msg, channel_id, chan.force_shutdown(), $self.get_channel_update(&chan).ok()))
                        },
+                       Err(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"); }
                }
        }
 }
@@ -448,6 +476,31 @@ macro_rules! try_chan_entry {
                                }
                                return Err(MsgHandleErrInternal::from_finish_shutdown(msg, channel_id, chan.force_shutdown(), $self.get_channel_update(&chan).ok()))
                        },
+                       Err(ChannelError::CloseDelayBroadcast { msg, update }) => {
+                               log_error!($self, "Channel {} need to be shutdown but closing transactions not broadcast due to {}", log_bytes!($entry.key()[..]), msg);
+                               let (channel_id, mut chan) = $entry.remove_entry();
+                               if let Some(short_id) = chan.get_short_channel_id() {
+                                       $channel_state.short_to_id.remove(&short_id);
+                               }
+                               if let Some(update) = update {
+                                       if let Err(e) = $self.monitor.add_update_monitor(update.get_funding_txo().unwrap(), update) {
+                                               match e {
+                                                       // Upstream channel is dead, but we want at least to fail backward HTLCs to save
+                                                       // downstream channels. In case of PermanentFailure, we are not going to be able
+                                                       // to claim back to_remote output on remote commitment transaction. Doesn't
+                                                       // make a difference here, we are concern about HTLCs circuit, not onchain funds.
+                                                       ChannelMonitorUpdateErr::PermanentFailure => {},
+                                                       ChannelMonitorUpdateErr::TemporaryFailure => {},
+                                               }
+                                       }
+                               }
+                               let mut shutdown_res = chan.force_shutdown();
+                               if shutdown_res.0.len() >= 1 {
+                                       log_error!($self, "You have a toxic local commitment transaction {} avaible in channel monitor, read comment in ChannelMonitor::get_latest_local_commitment_txn to be informed of manual action to take", shutdown_res.0[0].txid());
+                               }
+                               shutdown_res.0.clear();
+                               return Err(MsgHandleErrInternal::from_finish_shutdown(msg, channel_id, shutdown_res, $self.get_channel_update(&chan).ok()))
+                       }
                }
        }
 }
@@ -550,7 +603,6 @@ impl ChannelManager {
                        channel_state: Mutex::new(ChannelHolder{
                                by_id: HashMap::new(),
                                short_to_id: HashMap::new(),
-                               next_forward: Instant::now(),
                                forward_htlcs: HashMap::new(),
                                claimable_htlcs: HashMap::new(),
                                pending_msg_events: Vec::new(),
@@ -614,12 +666,16 @@ impl ChannelManager {
                let channel_state = self.channel_state.lock().unwrap();
                let mut res = Vec::with_capacity(channel_state.by_id.len());
                for (channel_id, channel) in channel_state.by_id.iter() {
+                       let (inbound_capacity_msat, outbound_capacity_msat) = channel.get_inbound_outbound_available_balance_msat();
                        res.push(ChannelDetails {
                                channel_id: (*channel_id).clone(),
                                short_channel_id: channel.get_short_channel_id(),
                                remote_network_id: channel.get_their_node_id(),
                                channel_value_satoshis: channel.get_value_satoshis(),
+                               inbound_capacity_msat,
+                               outbound_capacity_msat,
                                user_id: channel.get_user_id(),
+                               is_live: channel.is_live(),
                        });
                }
                res
@@ -627,6 +683,9 @@ impl ChannelManager {
 
        /// Gets the list of usable channels, in random order. Useful as an argument to
        /// Router::get_route to ensure non-announced channels are used.
+       ///
+       /// These are guaranteed to have their is_live value set to true, see the documentation for
+       /// ChannelDetails::is_live for more info on exactly what the criteria are.
        pub fn list_usable_channels(&self) -> Vec<ChannelDetails> {
                let channel_state = self.channel_state.lock().unwrap();
                let mut res = Vec::with_capacity(channel_state.by_id.len());
@@ -635,12 +694,16 @@ impl ChannelManager {
                        // internal/external nomenclature, but that's ok cause that's probably what the user
                        // really wanted anyway.
                        if channel.is_live() {
+                               let (inbound_capacity_msat, outbound_capacity_msat) = channel.get_inbound_outbound_available_balance_msat();
                                res.push(ChannelDetails {
                                        channel_id: (*channel_id).clone(),
                                        short_channel_id: channel.get_short_channel_id(),
                                        remote_network_id: channel.get_their_node_id(),
                                        channel_value_satoshis: channel.get_value_satoshis(),
+                                       inbound_capacity_msat,
+                                       outbound_capacity_msat,
                                        user_id: channel.get_user_id(),
+                                       is_live: true,
                                });
                        }
                }
@@ -946,9 +1009,9 @@ impl ChannelManager {
 
        /// 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, HandleError> {
+       fn get_channel_update(&self, chan: &Channel) -> Result<msgs::ChannelUpdate, LightningError> {
                let short_channel_id = match chan.get_short_channel_id() {
-                       None => return Err(HandleError{err: "Channel not yet established", action: None}),
+                       None => return Err(LightningError{err: "Channel not yet established", action: None}),
                        Some(id) => id,
                };
 
@@ -1103,7 +1166,7 @@ impl ChannelManager {
        pub fn funding_transaction_generated(&self, temporary_channel_id: &[u8; 32], funding_txo: OutPoint) {
                let _ = self.total_consistency_lock.read().unwrap();
 
-               let (chan, msg, chan_monitor) = {
+               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) {
@@ -1134,8 +1197,30 @@ impl ChannelManager {
                };
                // Because we have exclusive ownership of the channel here we can release the channel_state
                // lock before add_update_monitor
-               if let Err(_e) = self.monitor.add_update_monitor(chan_monitor.get_funding_txo().unwrap(), chan_monitor) {
-                       unimplemented!();
+               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!(),
+                                       }
+                               },
+                               ChannelMonitorUpdateErr::TemporaryFailure => {
+                                       // Its completely fine to continue with a FundingCreated until the monitor
+                                       // update is persisted, as long as we don't generate the FundingBroadcastSafe
+                                       // until the monitor has been safely persisted (as funding broadcast is not,
+                                       // in fact, safe).
+                                       chan.monitor_update_failed(false, false, Vec::new(), Vec::new());
+                               },
+                       }
                }
 
                let mut channel_state = self.channel_state.lock().unwrap();
@@ -1185,10 +1270,6 @@ impl ChannelManager {
                        let mut channel_state_lock = self.channel_state.lock().unwrap();
                        let channel_state = channel_state_lock.borrow_parts();
 
-                       if cfg!(not(feature = "fuzztarget")) && Instant::now() < *channel_state.next_forward {
-                               return;
-                       }
-
                        for (short_chan_id, mut pending_forwards) in channel_state.forward_htlcs.drain() {
                                if short_chan_id != 0 {
                                        let forward_chan_id = match channel_state.short_to_id.get(&short_chan_id) {
@@ -1407,7 +1488,7 @@ impl ChannelManager {
                                log_trace!(self, "Failing outbound payment HTLC with payment_hash {}", log_bytes!(payment_hash.0));
                                mem::drop(channel_state_lock);
                                match &onion_error {
-                                       &HTLCFailReason::ErrorPacket { ref err } => {
+                                       &HTLCFailReason::LightningError { ref err } => {
 #[cfg(test)]
                                                let (channel_update, payment_retryable, onion_error_code) = onion_utils::process_onion_failure(&self.secp_ctx, &self.logger, &source, err.data.clone());
 #[cfg(not(test))]
@@ -1460,16 +1541,15 @@ impl ChannelManager {
                                                let packet = onion_utils::build_failure_packet(&incoming_packet_shared_secret, failure_code, &data[..]).encode();
                                                onion_utils::encrypt_failure_packet(&incoming_packet_shared_secret, &packet)
                                        },
-                                       HTLCFailReason::ErrorPacket { err } => {
-                                               log_trace!(self, "Failing HTLC with payment_hash {} backwards with pre-built ErrorPacket", log_bytes!(payment_hash.0));
+                                       HTLCFailReason::LightningError { err } => {
+                                               log_trace!(self, "Failing HTLC with payment_hash {} backwards with pre-built LightningError", log_bytes!(payment_hash.0));
                                                onion_utils::encrypt_failure_packet(&incoming_packet_shared_secret, &err.data)
                                        }
                                };
 
                                let mut forward_event = None;
                                if channel_state_lock.forward_htlcs.is_empty() {
-                                       forward_event = Some(Instant::now() + Duration::from_millis(((rng::rand_f32() * 4.0 + 1.0) * MIN_HTLC_RELAY_HOLDING_CELL_MILLIS as f32) as u64));
-                                       channel_state_lock.next_forward = forward_event.unwrap();
+                                       forward_event = Some(Duration::from_millis(MIN_HTLC_RELAY_HOLDING_CELL_MILLIS));
                                }
                                match channel_state_lock.forward_htlcs.entry(short_channel_id) {
                                        hash_map::Entry::Occupied(mut entry) => {
@@ -1604,6 +1684,7 @@ impl ChannelManager {
                let mut close_results = Vec::new();
                let mut htlc_forwards = Vec::new();
                let mut htlc_failures = Vec::new();
+               let mut pending_events = Vec::new();
                let _ = self.total_consistency_lock.read().unwrap();
 
                {
@@ -1638,7 +1719,7 @@ impl ChannelManager {
                                                        ChannelMonitorUpdateErr::TemporaryFailure => true,
                                                }
                                        } else {
-                                               let (raa, commitment_update, order, pending_forwards, mut pending_failures) = channel.monitor_updating_restored();
+                                               let (raa, commitment_update, order, pending_forwards, mut pending_failures, needs_broadcast_safe, funding_locked) = channel.monitor_updating_restored();
                                                if !pending_forwards.is_empty() {
                                                        htlc_forwards.push((channel.get_short_channel_id().expect("We can't have pending forwards before funding confirmation"), pending_forwards));
                                                }
@@ -1670,12 +1751,33 @@ impl ChannelManager {
                                                                handle_cs!();
                                                        },
                                                }
+                                               if needs_broadcast_safe {
+                                                       pending_events.push(events::Event::FundingBroadcastSafe {
+                                                               funding_txo: channel.get_funding_txo().unwrap(),
+                                                               user_channel_id: channel.get_user_id(),
+                                                       });
+                                               }
+                                               if let Some(msg) = funding_locked {
+                                                       pending_msg_events.push(events::MessageSendEvent::SendFundingLocked {
+                                                               node_id: channel.get_their_node_id(),
+                                                               msg,
+                                                       });
+                                                       if let Some(announcement_sigs) = self.get_announcement_sigs(channel) {
+                                                               pending_msg_events.push(events::MessageSendEvent::SendAnnouncementSignatures {
+                                                                       node_id: channel.get_their_node_id(),
+                                                                       msg: announcement_sigs,
+                                                               });
+                                                       }
+                                                       short_to_id.insert(channel.get_short_channel_id().unwrap(), channel.channel_id());
+                                               }
                                                true
                                        }
                                } else { true }
                        });
                }
 
+               self.pending_events.lock().unwrap().append(&mut pending_events);
+
                for failure in htlc_failures.drain(..) {
                        self.fail_htlc_backwards_internal(self.channel_state.lock().unwrap(), failure.0, &failure.1, failure.2);
                }
@@ -1686,12 +1788,12 @@ impl ChannelManager {
                }
        }
 
-       fn internal_open_channel(&self, their_node_id: &PublicKey, msg: &msgs::OpenChannel) -> Result<(), MsgHandleErrInternal> {
+       fn internal_open_channel(&self, their_node_id: &PublicKey, their_local_features: LocalFeatures, 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(), 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_local_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();
@@ -1708,7 +1810,7 @@ impl ChannelManager {
                Ok(())
        }
 
-       fn internal_accept_channel(&self, their_node_id: &PublicKey, msg: &msgs::AcceptChannel) -> Result<(), MsgHandleErrInternal> {
+       fn internal_accept_channel(&self, their_node_id: &PublicKey, their_local_features: LocalFeatures, 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();
@@ -1718,7 +1820,7 @@ impl ChannelManager {
                                                //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), channel_state, chan);
+                                       try_chan_entry!(self, chan.get_mut().accept_channel(&msg, &self.default_configuration, their_local_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
@@ -1736,7 +1838,7 @@ impl ChannelManager {
        }
 
        fn internal_funding_created(&self, their_node_id: &PublicKey, msg: &msgs::FundingCreated) -> Result<(), MsgHandleErrInternal> {
-               let ((funding_msg, monitor_update), chan) = {
+               let ((funding_msg, monitor_update), mut chan) = {
                        let mut channel_lock = self.channel_state.lock().unwrap();
                        let channel_state = channel_lock.borrow_parts();
                        match channel_state.by_id.entry(msg.temporary_channel_id.clone()) {
@@ -1752,8 +1854,23 @@ impl ChannelManager {
                };
                // Because we have exclusive ownership of the channel here we can release the channel_state
                // lock before add_update_monitor
-               if let Err(_e) = self.monitor.add_update_monitor(monitor_update.get_funding_txo().unwrap(), monitor_update) {
-                       unimplemented!();
+               if let Err(e) = self.monitor.add_update_monitor(monitor_update.get_funding_txo().unwrap(), monitor_update) {
+                       match e {
+                               ChannelMonitorUpdateErr::PermanentFailure => {
+                                       // Note that we reply with the new channel_id in error messages if we gave up on the
+                                       // channel, not the temporary_channel_id. This is compatible with ourselves, but the
+                                       // spec is somewhat ambiguous here. Not a huge deal since we'll send error messages for
+                                       // any messages referencing a previously-closed channel anyway.
+                                       return Err(MsgHandleErrInternal::from_finish_shutdown("ChannelMonitor storage failure", funding_msg.channel_id, chan.force_shutdown(), None));
+                               },
+                               ChannelMonitorUpdateErr::TemporaryFailure => {
+                                       // There's no problem signing a counterparty's funding transaction if our monitor
+                                       // hasn't persisted to disk yet - we can't lose money on a transaction that we haven't
+                                       // accepted payment from yet. We do, however, need to wait to send our funding_locked
+                                       // until we have persisted our monitor.
+                                       chan.monitor_update_failed(false, false, Vec::new(), Vec::new());
+                               },
+                       }
                }
                let mut channel_state_lock = self.channel_state.lock().unwrap();
                let channel_state = channel_state_lock.borrow_parts();
@@ -1783,8 +1900,8 @@ impl ChannelManager {
                                                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);
-                                       if let Err(_e) = self.monitor.add_update_monitor(chan_monitor.get_funding_txo().unwrap(), chan_monitor) {
-                                               unimplemented!();
+                                       if let Err(e) = self.monitor.add_update_monitor(chan_monitor.get_funding_txo().unwrap(), chan_monitor) {
+                                               return_monitor_err!(self, e, channel_state, chan, RAACommitmentOrder::RevokeAndACKFirst, false, false);
                                        }
                                        (chan.get().get_funding_txo().unwrap(), chan.get().get_user_id())
                                },
@@ -2001,7 +2118,7 @@ impl ChannelManager {
                                        //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::ErrorPacket { err: msg.reason.clone() }), channel_state, chan);
+                               try_chan_entry!(self, chan.get_mut().update_fail_htlc(&msg, HTLCFailReason::LightningError { err: msg.reason.clone() }), channel_state, chan);
                        },
                        hash_map::Entry::Vacant(_) => return Err(MsgHandleErrInternal::send_err_msg_no_close("Failed to find corresponding channel", msg.channel_id))
                }
@@ -2078,8 +2195,7 @@ impl ChannelManager {
                        if !pending_forwards.is_empty() {
                                let mut channel_state = self.channel_state.lock().unwrap();
                                if channel_state.forward_htlcs.is_empty() {
-                                       forward_event = Some(Instant::now() + Duration::from_millis(((rng::rand_f32() * 4.0 + 1.0) * MIN_HTLC_RELAY_HOLDING_CELL_MILLIS as f32) as u64));
-                                       channel_state.next_forward = forward_event.unwrap();
+                                       forward_event = Some(Duration::from_millis(MIN_HTLC_RELAY_HOLDING_CELL_MILLIS))
                                }
                                for (forward_info, prev_htlc_id) in pending_forwards.drain(..) {
                                        match channel_state.forward_htlcs.entry(forward_info.short_channel_id) {
@@ -2176,7 +2292,7 @@ impl ChannelManager {
                                        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() {
-                                       return Err(MsgHandleErrInternal::from_no_close(HandleError{err: "Got an announcement_signatures before we were ready for it", action: Some(msgs::ErrorAction::IgnoreError)}));
+                                       return Err(MsgHandleErrInternal::from_no_close(LightningError{err: "Got an announcement_signatures before we were ready for it", action: Some(msgs::ErrorAction::IgnoreError)}));
                                }
 
                                let our_node_id = self.get_our_node_id();
@@ -2510,82 +2626,82 @@ impl ChainListener for ChannelManager {
 
 impl ChannelMessageHandler for ChannelManager {
        //TODO: Handle errors and close channel (or so)
-       fn handle_open_channel(&self, their_node_id: &PublicKey, msg: &msgs::OpenChannel) -> Result<(), HandleError> {
+       fn handle_open_channel(&self, their_node_id: &PublicKey, their_local_features: LocalFeatures, msg: &msgs::OpenChannel) -> Result<(), LightningError> {
                let _ = self.total_consistency_lock.read().unwrap();
-               handle_error!(self, self.internal_open_channel(their_node_id, msg))
+               handle_error!(self, self.internal_open_channel(their_node_id, their_local_features, msg))
        }
 
-       fn handle_accept_channel(&self, their_node_id: &PublicKey, msg: &msgs::AcceptChannel) -> Result<(), HandleError> {
+       fn handle_accept_channel(&self, their_node_id: &PublicKey, their_local_features: LocalFeatures, msg: &msgs::AcceptChannel) -> Result<(), LightningError> {
                let _ = self.total_consistency_lock.read().unwrap();
-               handle_error!(self, self.internal_accept_channel(their_node_id, msg))
+               handle_error!(self, self.internal_accept_channel(their_node_id, their_local_features, msg))
        }
 
-       fn handle_funding_created(&self, their_node_id: &PublicKey, msg: &msgs::FundingCreated) -> Result<(), HandleError> {
+       fn handle_funding_created(&self, their_node_id: &PublicKey, msg: &msgs::FundingCreated) -> Result<(), LightningError> {
                let _ = self.total_consistency_lock.read().unwrap();
                handle_error!(self, self.internal_funding_created(their_node_id, msg))
        }
 
-       fn handle_funding_signed(&self, their_node_id: &PublicKey, msg: &msgs::FundingSigned) -> Result<(), HandleError> {
+       fn handle_funding_signed(&self, their_node_id: &PublicKey, msg: &msgs::FundingSigned) -> Result<(), LightningError> {
                let _ = self.total_consistency_lock.read().unwrap();
                handle_error!(self, self.internal_funding_signed(their_node_id, msg))
        }
 
-       fn handle_funding_locked(&self, their_node_id: &PublicKey, msg: &msgs::FundingLocked) -> Result<(), HandleError> {
+       fn handle_funding_locked(&self, their_node_id: &PublicKey, msg: &msgs::FundingLocked) -> Result<(), LightningError> {
                let _ = self.total_consistency_lock.read().unwrap();
                handle_error!(self, self.internal_funding_locked(their_node_id, msg))
        }
 
-       fn handle_shutdown(&self, their_node_id: &PublicKey, msg: &msgs::Shutdown) -> Result<(), HandleError> {
+       fn handle_shutdown(&self, their_node_id: &PublicKey, msg: &msgs::Shutdown) -> Result<(), LightningError> {
                let _ = self.total_consistency_lock.read().unwrap();
                handle_error!(self, self.internal_shutdown(their_node_id, msg))
        }
 
-       fn handle_closing_signed(&self, their_node_id: &PublicKey, msg: &msgs::ClosingSigned) -> Result<(), HandleError> {
+       fn handle_closing_signed(&self, their_node_id: &PublicKey, msg: &msgs::ClosingSigned) -> Result<(), LightningError> {
                let _ = self.total_consistency_lock.read().unwrap();
                handle_error!(self, self.internal_closing_signed(their_node_id, msg))
        }
 
-       fn handle_update_add_htlc(&self, their_node_id: &PublicKey, msg: &msgs::UpdateAddHTLC) -> Result<(), msgs::HandleError> {
+       fn handle_update_add_htlc(&self, their_node_id: &PublicKey, msg: &msgs::UpdateAddHTLC) -> Result<(), LightningError> {
                let _ = self.total_consistency_lock.read().unwrap();
                handle_error!(self, self.internal_update_add_htlc(their_node_id, msg))
        }
 
-       fn handle_update_fulfill_htlc(&self, their_node_id: &PublicKey, msg: &msgs::UpdateFulfillHTLC) -> Result<(), HandleError> {
+       fn handle_update_fulfill_htlc(&self, their_node_id: &PublicKey, msg: &msgs::UpdateFulfillHTLC) -> Result<(), LightningError> {
                let _ = self.total_consistency_lock.read().unwrap();
                handle_error!(self, self.internal_update_fulfill_htlc(their_node_id, msg))
        }
 
-       fn handle_update_fail_htlc(&self, their_node_id: &PublicKey, msg: &msgs::UpdateFailHTLC) -> Result<(), HandleError> {
+       fn handle_update_fail_htlc(&self, their_node_id: &PublicKey, msg: &msgs::UpdateFailHTLC) -> Result<(), LightningError> {
                let _ = self.total_consistency_lock.read().unwrap();
                handle_error!(self, self.internal_update_fail_htlc(their_node_id, msg))
        }
 
-       fn handle_update_fail_malformed_htlc(&self, their_node_id: &PublicKey, msg: &msgs::UpdateFailMalformedHTLC) -> Result<(), HandleError> {
+       fn handle_update_fail_malformed_htlc(&self, their_node_id: &PublicKey, msg: &msgs::UpdateFailMalformedHTLC) -> Result<(), LightningError> {
                let _ = self.total_consistency_lock.read().unwrap();
                handle_error!(self, self.internal_update_fail_malformed_htlc(their_node_id, msg))
        }
 
-       fn handle_commitment_signed(&self, their_node_id: &PublicKey, msg: &msgs::CommitmentSigned) -> Result<(), HandleError> {
+       fn handle_commitment_signed(&self, their_node_id: &PublicKey, msg: &msgs::CommitmentSigned) -> Result<(), LightningError> {
                let _ = self.total_consistency_lock.read().unwrap();
                handle_error!(self, self.internal_commitment_signed(their_node_id, msg))
        }
 
-       fn handle_revoke_and_ack(&self, their_node_id: &PublicKey, msg: &msgs::RevokeAndACK) -> Result<(), HandleError> {
+       fn handle_revoke_and_ack(&self, their_node_id: &PublicKey, msg: &msgs::RevokeAndACK) -> Result<(), LightningError> {
                let _ = self.total_consistency_lock.read().unwrap();
                handle_error!(self, self.internal_revoke_and_ack(their_node_id, msg))
        }
 
-       fn handle_update_fee(&self, their_node_id: &PublicKey, msg: &msgs::UpdateFee) -> Result<(), HandleError> {
+       fn handle_update_fee(&self, their_node_id: &PublicKey, msg: &msgs::UpdateFee) -> Result<(), LightningError> {
                let _ = self.total_consistency_lock.read().unwrap();
                handle_error!(self, self.internal_update_fee(their_node_id, msg))
        }
 
-       fn handle_announcement_signatures(&self, their_node_id: &PublicKey, msg: &msgs::AnnouncementSignatures) -> Result<(), HandleError> {
+       fn handle_announcement_signatures(&self, their_node_id: &PublicKey, msg: &msgs::AnnouncementSignatures) -> Result<(), LightningError> {
                let _ = self.total_consistency_lock.read().unwrap();
                handle_error!(self, self.internal_announcement_signatures(their_node_id, msg))
        }
 
-       fn handle_channel_reestablish(&self, their_node_id: &PublicKey, msg: &msgs::ChannelReestablish) -> Result<(), HandleError> {
+       fn handle_channel_reestablish(&self, their_node_id: &PublicKey, msg: &msgs::ChannelReestablish) -> Result<(), LightningError> {
                let _ = self.total_consistency_lock.read().unwrap();
                handle_error!(self, self.internal_channel_reestablish(their_node_id, msg))
        }
@@ -2830,7 +2946,7 @@ impl<R: ::std::io::Read> Readable<R> for HTLCSource {
 impl Writeable for HTLCFailReason {
        fn write<W: Writer>(&self, writer: &mut W) -> Result<(), ::std::io::Error> {
                match self {
-                       &HTLCFailReason::ErrorPacket { ref err } => {
+                       &HTLCFailReason::LightningError { ref err } => {
                                0u8.write(writer)?;
                                err.write(writer)?;
                        },
@@ -2847,7 +2963,7 @@ impl Writeable for HTLCFailReason {
 impl<R: ::std::io::Read> Readable<R> for HTLCFailReason {
        fn read(reader: &mut R) -> Result<HTLCFailReason, DecodeError> {
                match <u8 as Readable<R>>::read(reader)? {
-                       0 => Ok(HTLCFailReason::ErrorPacket { err: Readable::read(reader)? }),
+                       0 => Ok(HTLCFailReason::LightningError { err: Readable::read(reader)? }),
                        1 => Ok(HTLCFailReason::Reason {
                                failure_code: Readable::read(reader)?,
                                data: Readable::read(reader)?,
@@ -3088,7 +3204,6 @@ impl<'a, R : ::std::io::Read> ReadableArgs<R, ChannelManagerReadArgs<'a>> for (S
                        channel_state: Mutex::new(ChannelHolder {
                                by_id,
                                short_to_id,
-                               next_forward: Instant::now(),
                                forward_htlcs,
                                claimable_htlcs,
                                pending_msg_events: Vec::new(),