Remove fuzz channel_target.
[rust-lightning] / src / ln / channelmanager.rs
index af27a3f32134bd13a26805b887e8bc2758cb3b58..d6246166bb7c3ac9ddd6f2bc2aa6d1a094750230 100644 (file)
@@ -16,9 +16,10 @@ use ln::channel::{Channel, ChannelKeys};
 use ln::channelmonitor::ManyChannelMonitor;
 use ln::router::{Route,RouteHop};
 use ln::msgs;
-use ln::msgs::{HandleError,ChannelMessageHandler,MsgEncodable,MsgDecodable};
+use ln::msgs::{HandleError,ChannelMessageHandler};
 use util::{byte_utils, events, internal_traits, rng};
 use util::sha2::Sha256;
+use util::ser::{Readable, Writeable};
 use util::chacha20poly1305rfc::ChaCha20;
 use util::logger::Logger;
 use util::errors::APIError;
@@ -32,6 +33,7 @@ use crypto::symmetriccipher::SynchronousStreamCipher;
 use std::{ptr, mem};
 use std::collections::HashMap;
 use std::collections::hash_map;
+use std::io::Cursor;
 use std::sync::{Mutex,MutexGuard,Arc};
 use std::sync::atomic::{AtomicUsize, Ordering};
 use std::time::{Instant,Duration};
@@ -78,23 +80,6 @@ mod channel_held_info {
                Fail(HTLCFailureMsg),
        }
 
-       #[cfg(feature = "fuzztarget")]
-       impl PendingHTLCStatus {
-               pub fn dummy() -> Self {
-                       let secp_ctx = ::secp256k1::Secp256k1::signing_only();
-                       PendingHTLCStatus::Forward(PendingForwardHTLCInfo {
-                               onion_packet: None,
-                               incoming_shared_secret: SharedSecret::new(&secp_ctx,
-                                               &::secp256k1::key::PublicKey::from_secret_key(&secp_ctx, &SecretKey::from_slice(&secp_ctx, &[1; 32]).unwrap()),
-                                               &SecretKey::from_slice(&secp_ctx, &[1; 32]).unwrap()),
-                               payment_hash: [0; 32],
-                               short_channel_id: 0,
-                               amt_to_forward: 0,
-                               outgoing_cltv_value: 0,
-                       })
-               }
-       }
-
        /// Tracks the inbound corresponding to an outbound HTLC
        #[derive(Clone)]
        pub struct HTLCPreviousHopData {
@@ -112,7 +97,7 @@ mod channel_held_info {
                        session_priv: SecretKey,
                },
        }
-       #[cfg(any(test, feature = "fuzztarget"))]
+       #[cfg(test)]
        impl HTLCSource {
                pub fn dummy() -> Self {
                        HTLCSource::OutboundRoute {
@@ -123,7 +108,7 @@ mod channel_held_info {
        }
 
        #[derive(Clone)] // See Channel::revoke_and_ack for why, tl;dr: Rust bug
-       pub enum HTLCFailReason {
+       pub(crate) enum HTLCFailReason {
                ErrorPacket {
                        err: msgs::OnionErrorPacket,
                },
@@ -132,20 +117,8 @@ mod channel_held_info {
                        data: Vec<u8>,
                }
        }
-
-       #[cfg(feature = "fuzztarget")]
-       impl HTLCFailReason {
-               pub fn dummy() -> Self {
-                       HTLCFailReason::Reason {
-                               failure_code: 0, data: Vec::new(),
-                       }
-               }
-       }
 }
-#[cfg(feature = "fuzztarget")]
-pub use self::channel_held_info::*;
-#[cfg(not(feature = "fuzztarget"))]
-pub(crate) use self::channel_held_info::*;
+pub(super) use self::channel_held_info::*;
 
 struct MsgHandleErrInternal {
        err: msgs::HandleError,
@@ -819,7 +792,7 @@ impl ChannelManager {
                let next_hop_data = {
                        let mut decoded = [0; 65];
                        chacha.process(&msg.onion_routing_packet.hop_data[0..65], &mut decoded);
-                       match msgs::OnionHopData::decode(&decoded[..]) {
+                       match msgs::OnionHopData::read(&mut Cursor::new(&decoded[..])) {
                                Err(err) => {
                                        let error_code = match err {
                                                msgs::DecodeError::UnknownRealmByte => 0x4000 | 1,
@@ -1218,7 +1191,7 @@ impl ChannelManager {
                                                };
                                                match channel_state.claimable_htlcs.entry(forward_info.payment_hash) {
                                                        hash_map::Entry::Occupied(mut entry) => entry.get_mut().push(prev_hop_data),
-                                                       hash_map::Entry::Vacant(mut entry) => { entry.insert(vec![prev_hop_data]); },
+                                                       hash_map::Entry::Vacant(entry) => { entry.insert(vec![prev_hop_data]); },
                                                };
                                                new_events.push((None, events::Event::PaymentReceived {
                                                        payment_hash: forward_info.payment_hash,
@@ -1717,7 +1690,7 @@ impl ChannelManager {
                                        chacha.process(&packet_decrypted, &mut decryption_tmp[..]);
                                        packet_decrypted = decryption_tmp;
 
-                                       if let Ok(err_packet) = msgs::DecodedOnionErrorPacket::decode(&packet_decrypted) {
+                                       if let Ok(err_packet) = msgs::DecodedOnionErrorPacket::read(&mut Cursor::new(&packet_decrypted)) {
                                                if err_packet.failuremsg.len() >= 2 {
                                                        let um = ChannelManager::gen_um_from_shared_secret(&shared_secret);
 
@@ -1733,7 +1706,7 @@ impl ChannelManager {
                                                                                if err_packet.failuremsg.len() >= 4 {
                                                                                        let update_len = byte_utils::slice_to_be16(&err_packet.failuremsg[2..4]) as usize;
                                                                                        if err_packet.failuremsg.len() >= 4 + update_len {
-                                                                                               if let Ok(chan_update) = msgs::ChannelUpdate::decode(&err_packet.failuremsg[4..4 + update_len]) {
+                                                                                               if let Ok(chan_update) = msgs::ChannelUpdate::read(&mut Cursor::new(&err_packet.failuremsg[4..4 + update_len])) {
                                                                                                        res = Some(msgs::HTLCFailChannelUpdate::ChannelUpdateMessage {
                                                                                                                msg: chan_update,
                                                                                                        });
@@ -1902,7 +1875,27 @@ impl ChannelManager {
                Ok(())
        }
 
-
+       fn internal_channel_reestablish(&self, their_node_id: &PublicKey, msg: &msgs::ChannelReestablish) -> Result<(Option<msgs::FundingLocked>, Option<msgs::RevokeAndACK>, Option<msgs::CommitmentUpdate>), MsgHandleErrInternal> {
+               let (res, chan_monitor) = {
+                       let mut channel_state = self.channel_state.lock().unwrap();
+                       match channel_state.by_id.get_mut(&msg.channel_id) {
+                               Some(chan) => {
+                                       if chan.get_their_node_id() != *their_node_id {
+                                               return Err(MsgHandleErrInternal::send_err_msg_no_close("Got a message for a channel from the wrong node!", msg.channel_id));
+                                       }
+                                       let (funding_locked, revoke_and_ack, commitment_update, channel_monitor) = chan.channel_reestablish(msg).map_err(|e| MsgHandleErrInternal::from_maybe_close(e))?;
+                                       (Ok((funding_locked, revoke_and_ack, commitment_update)), channel_monitor)
+                               },
+                               None => return Err(MsgHandleErrInternal::send_err_msg_no_close("Failed to find corresponding channel", msg.channel_id))
+                       }
+               };
+               if let Some(monitor) = chan_monitor {
+                       if let Err(_e) = self.monitor.add_update_monitor(monitor.get_funding_txo().unwrap(), monitor) {
+                               unimplemented!();
+                       }
+               }
+               res
+       }
 }
 
 impl events::EventsProvider for ChannelManager {
@@ -2123,9 +2116,14 @@ impl ChannelMessageHandler for ChannelManager {
                handle_error!(self, self.internal_announcement_signatures(their_node_id, msg), their_node_id)
        }
 
+       fn handle_channel_reestablish(&self, their_node_id: &PublicKey, msg: &msgs::ChannelReestablish) -> Result<(Option<msgs::FundingLocked>, Option<msgs::RevokeAndACK>, Option<msgs::CommitmentUpdate>), HandleError> {
+               handle_error!(self, self.internal_channel_reestablish(their_node_id, msg), their_node_id)
+       }
+
        fn peer_disconnected(&self, their_node_id: &PublicKey, no_connection_possible: bool) {
                let mut new_events = Vec::new();
                let mut failed_channels = Vec::new();
+               let mut failed_payments = Vec::new();
                {
                        let mut channel_state_lock = self.channel_state.lock().unwrap();
                        let channel_state = channel_state_lock.borrow_parts();
@@ -2148,13 +2146,23 @@ impl ChannelMessageHandler for ChannelManager {
                                        }
                                });
                        } else {
-                               for chan in channel_state.by_id {
-                                       if chan.1.get_their_node_id() == *their_node_id {
-                                               //TODO: mark channel disabled (and maybe announce such after a timeout). Also
-                                               //fail and wipe any uncommitted outbound HTLCs as those are considered after
-                                               //reconnect.
+                               channel_state.by_id.retain(|_, chan| {
+                                       if chan.get_their_node_id() == *their_node_id {
+                                               //TODO: mark channel disabled (and maybe announce such after a timeout).
+                                               let failed_adds = chan.remove_uncommitted_htlcs_and_mark_paused();
+                                               if !failed_adds.is_empty() {
+                                                       let chan_update = self.get_channel_update(&chan).map(|u| u.encode_with_len()).unwrap(); // Cannot add/recv HTLCs before we have a short_id so unwrap is safe
+                                                       failed_payments.push((chan_update, failed_adds));
+                                               }
+                                               if chan.is_shutdown() {
+                                                       if let Some(short_id) = chan.get_short_channel_id() {
+                                                               short_to_id.remove(&short_id);
+                                                       }
+                                                       return false;
+                                               }
                                        }
-                               }
+                                       true
+                               })
                        }
                }
                for failure in failed_channels.drain(..) {
@@ -2166,6 +2174,32 @@ impl ChannelMessageHandler for ChannelManager {
                                pending_events.push(event);
                        }
                }
+               for (chan_update, mut htlc_sources) in failed_payments {
+                       for (htlc_source, payment_hash) in htlc_sources.drain(..) {
+                               self.fail_htlc_backwards_internal(self.channel_state.lock().unwrap(), htlc_source, &payment_hash, HTLCFailReason::Reason { failure_code: 0x1000 | 7, data: chan_update.clone() });
+                       }
+               }
+       }
+
+       fn peer_connected(&self, their_node_id: &PublicKey) -> Vec<msgs::ChannelReestablish> {
+               let mut res = Vec::new();
+               let mut channel_state = self.channel_state.lock().unwrap();
+               channel_state.by_id.retain(|_, chan| {
+                       if chan.get_their_node_id() == *their_node_id {
+                               if !chan.have_received_message() {
+                                       // If we created this (outbound) channel while we were disconnected from the
+                                       // peer we probably failed to send the open_channel message, which is now
+                                       // lost. We can't have had anything pending related to this channel, so we just
+                                       // drop it.
+                                       false
+                               } else {
+                                       res.push(chan.get_channel_reestablish());
+                                       true
+                               }
+                       } else { true }
+               });
+               //TODO: Also re-broadcast announcement_signatures
+               res
        }
 
        fn handle_error(&self, their_node_id: &PublicKey, msg: &msgs::ErrorMessage) {
@@ -2189,16 +2223,16 @@ mod tests {
        use ln::channelmanager::{ChannelManager,OnionKeys};
        use ln::router::{Route, RouteHop, Router};
        use ln::msgs;
-       use ln::msgs::{MsgEncodable,ChannelMessageHandler,RoutingMessageHandler};
+       use ln::msgs::{ChannelMessageHandler,RoutingMessageHandler};
        use util::test_utils;
        use util::events::{Event, EventsProvider};
-       use util::logger::Logger;
        use util::errors::APIError;
+       use util::logger::Logger;
+       use util::ser::Writeable;
 
        use bitcoin::util::hash::Sha256dHash;
        use bitcoin::blockdata::block::{Block, BlockHeader};
        use bitcoin::blockdata::transaction::{Transaction, TxOut};
-       use bitcoin::blockdata::transaction::OutPoint as BitcoinOutPoint;
        use bitcoin::blockdata::constants::genesis_block;
        use bitcoin::network::constants::Network;
        use bitcoin::network::serialize::serialize;
@@ -2391,6 +2425,13 @@ mod tests {
                network_payment_count: Rc<RefCell<u8>>,
                network_chan_count: Rc<RefCell<u32>>,
        }
+       impl Drop for Node {
+               fn drop(&mut self) {
+                       // Check that we processed all pending events
+                       assert_eq!(self.node.get_and_clear_pending_events().len(), 0);
+                       assert_eq!(self.chan_monitor.added_monitors.lock().unwrap().len(), 0);
+               }
+       }
 
        fn create_chan_between_nodes(node_a: &Node, node_b: &Node) -> (msgs::ChannelAnnouncement, msgs::ChannelUpdate, msgs::ChannelUpdate, [u8; 32], Transaction) {
                node_a.node.create_channel(node_b.node.get_our_node_id(), 100000, 10001, 42).unwrap();
@@ -2735,7 +2776,7 @@ mod tests {
                (our_payment_preimage, our_payment_hash)
        }
 
-       fn claim_payment(origin_node: &Node, expected_route: &[&Node], our_payment_preimage: [u8; 32]) {
+       fn claim_payment_along_route(origin_node: &Node, expected_route: &[&Node], skip_last: bool, our_payment_preimage: [u8; 32]) {
                assert!(expected_route.last().unwrap().node.claim_funds(our_payment_preimage));
                {
                        let mut added_monitors = expected_route.last().unwrap().chan_monitor.added_monitors.lock().unwrap();
@@ -2764,40 +2805,51 @@ mod tests {
 
                let mut expected_next_node = expected_route.last().unwrap().node.get_our_node_id();
                let mut prev_node = expected_route.last().unwrap();
-               for node in expected_route.iter().rev() {
+               for (idx, node) in expected_route.iter().rev().enumerate() {
                        assert_eq!(expected_next_node, node.node.get_our_node_id());
                        if next_msgs.is_some() {
                                update_fulfill_dance!(node, prev_node, false);
                        }
 
                        let events = node.node.get_and_clear_pending_events();
+                       if !skip_last || idx != expected_route.len() - 1 {
+                               assert_eq!(events.len(), 1);
+                               match events[0] {
+                                       Event::UpdateHTLCs { ref node_id, updates: msgs::CommitmentUpdate { ref update_add_htlcs, ref update_fulfill_htlcs, ref update_fail_htlcs, ref update_fail_malformed_htlcs, ref commitment_signed } } => {
+                                               assert!(update_add_htlcs.is_empty());
+                                               assert_eq!(update_fulfill_htlcs.len(), 1);
+                                               assert!(update_fail_htlcs.is_empty());
+                                               assert!(update_fail_malformed_htlcs.is_empty());
+                                               expected_next_node = node_id.clone();
+                                               next_msgs = Some((update_fulfill_htlcs[0].clone(), commitment_signed.clone()));
+                                       },
+                                       _ => panic!("Unexpected event"),
+                               }
+                       } else {
+                               assert!(events.is_empty());
+                       }
+                       if !skip_last && idx == expected_route.len() - 1 {
+                               assert_eq!(expected_next_node, origin_node.node.get_our_node_id());
+                       }
+
+                       prev_node = node;
+               }
+
+               if !skip_last {
+                       update_fulfill_dance!(origin_node, expected_route.first().unwrap(), true);
+                       let events = origin_node.node.get_and_clear_pending_events();
                        assert_eq!(events.len(), 1);
                        match events[0] {
-                               Event::UpdateHTLCs { ref node_id, updates: msgs::CommitmentUpdate { ref update_add_htlcs, ref update_fulfill_htlcs, ref update_fail_htlcs, ref update_fail_malformed_htlcs, ref commitment_signed } } => {
-                                       assert!(update_add_htlcs.is_empty());
-                                       assert_eq!(update_fulfill_htlcs.len(), 1);
-                                       assert!(update_fail_htlcs.is_empty());
-                                       assert!(update_fail_malformed_htlcs.is_empty());
-                                       expected_next_node = node_id.clone();
-                                       next_msgs = Some((update_fulfill_htlcs[0].clone(), commitment_signed.clone()));
+                               Event::PaymentSent { payment_preimage } => {
+                                       assert_eq!(payment_preimage, our_payment_preimage);
                                },
                                _ => panic!("Unexpected event"),
-                       };
-
-                       prev_node = node;
+                       }
                }
+       }
 
-               assert_eq!(expected_next_node, origin_node.node.get_our_node_id());
-               update_fulfill_dance!(origin_node, expected_route.first().unwrap(), true);
-
-               let events = origin_node.node.get_and_clear_pending_events();
-               assert_eq!(events.len(), 1);
-               match events[0] {
-                       Event::PaymentSent { payment_preimage } => {
-                               assert_eq!(payment_preimage, our_payment_preimage);
-                       },
-                       _ => panic!("Unexpected event"),
-               }
+       fn claim_payment(origin_node: &Node, expected_route: &[&Node], our_payment_preimage: [u8; 32]) {
+               claim_payment_along_route(origin_node, expected_route, false, our_payment_preimage);
        }
 
        const TEST_FINAL_CLTV: u32 = 32;
@@ -2841,7 +2893,7 @@ mod tests {
                claim_payment(&origin, expected_route, our_payment_preimage);
        }
 
-       fn fail_payment(origin_node: &Node, expected_route: &[&Node], our_payment_hash: [u8; 32]) {
+       fn fail_payment_along_route(origin_node: &Node, expected_route: &[&Node], skip_last: bool, our_payment_hash: [u8; 32]) {
                assert!(expected_route.last().unwrap().node.fail_htlc_backwards(&our_payment_hash));
                {
                        let mut added_monitors = expected_route.last().unwrap().chan_monitor.added_monitors.lock().unwrap();
@@ -2861,42 +2913,57 @@ mod tests {
 
                let mut expected_next_node = expected_route.last().unwrap().node.get_our_node_id();
                let mut prev_node = expected_route.last().unwrap();
-               for node in expected_route.iter().rev() {
+               for (idx, node) in expected_route.iter().rev().enumerate() {
                        assert_eq!(expected_next_node, node.node.get_our_node_id());
                        if next_msgs.is_some() {
-                               update_fail_dance!(node, prev_node, false);
+                               // We may be the "last node" for the purpose of the commitment dance if we're
+                               // skipping the last node (implying it is disconnected) and we're the
+                               // second-to-last node!
+                               update_fail_dance!(node, prev_node, skip_last && idx == expected_route.len() - 1);
                        }
 
                        let events = node.node.get_and_clear_pending_events();
-                       assert_eq!(events.len(), 1);
-                       match events[0] {
-                               Event::UpdateHTLCs { ref node_id, updates: msgs::CommitmentUpdate { ref update_add_htlcs, ref update_fulfill_htlcs, ref update_fail_htlcs, ref update_fail_malformed_htlcs, ref commitment_signed } } => {
-                                       assert!(update_add_htlcs.is_empty());
-                                       assert!(update_fulfill_htlcs.is_empty());
-                                       assert_eq!(update_fail_htlcs.len(), 1);
-                                       assert!(update_fail_malformed_htlcs.is_empty());
-                                       expected_next_node = node_id.clone();
-                                       next_msgs = Some((update_fail_htlcs[0].clone(), commitment_signed.clone()));
-                               },
-                               _ => panic!("Unexpected event"),
-                       };
+                       if !skip_last || idx != expected_route.len() - 1 {
+                               assert_eq!(events.len(), 1);
+                               match events[0] {
+                                       Event::UpdateHTLCs { ref node_id, updates: msgs::CommitmentUpdate { ref update_add_htlcs, ref update_fulfill_htlcs, ref update_fail_htlcs, ref update_fail_malformed_htlcs, ref commitment_signed } } => {
+                                               assert!(update_add_htlcs.is_empty());
+                                               assert!(update_fulfill_htlcs.is_empty());
+                                               assert_eq!(update_fail_htlcs.len(), 1);
+                                               assert!(update_fail_malformed_htlcs.is_empty());
+                                               expected_next_node = node_id.clone();
+                                               next_msgs = Some((update_fail_htlcs[0].clone(), commitment_signed.clone()));
+                                       },
+                                       _ => panic!("Unexpected event"),
+                               }
+                       } else {
+                               assert!(events.is_empty());
+                       }
+                       if !skip_last && idx == expected_route.len() - 1 {
+                               assert_eq!(expected_next_node, origin_node.node.get_our_node_id());
+                       }
 
                        prev_node = node;
                }
 
-               assert_eq!(expected_next_node, origin_node.node.get_our_node_id());
-               update_fail_dance!(origin_node, expected_route.first().unwrap(), true);
+               if !skip_last {
+                       update_fail_dance!(origin_node, expected_route.first().unwrap(), true);
 
-               let events = origin_node.node.get_and_clear_pending_events();
-               assert_eq!(events.len(), 1);
-               match events[0] {
-                       Event::PaymentFailed { payment_hash } => {
-                               assert_eq!(payment_hash, our_payment_hash);
-                       },
-                       _ => panic!("Unexpected event"),
+                       let events = origin_node.node.get_and_clear_pending_events();
+                       assert_eq!(events.len(), 1);
+                       match events[0] {
+                               Event::PaymentFailed { payment_hash } => {
+                                       assert_eq!(payment_hash, our_payment_hash);
+                               },
+                               _ => panic!("Unexpected event"),
+                       }
                }
        }
 
+       fn fail_payment(origin_node: &Node, expected_route: &[&Node], our_payment_hash: [u8; 32]) {
+               fail_payment_along_route(origin_node, expected_route, false, our_payment_hash);
+       }
+
        fn create_network(node_count: usize) -> Vec<Node> {
                let mut nodes = Vec::new();
                let mut rng = thread_rng();
@@ -3037,12 +3104,6 @@ mod tests {
                close_channel(&nodes[2], &nodes[3], &chan_3.2, chan_3.3, true);
                close_channel(&nodes[1], &nodes[3], &chan_4.2, chan_4.3, false);
                close_channel(&nodes[1], &nodes[3], &chan_5.2, chan_5.3, false);
-
-               // Check that we processed all pending events
-               for node in nodes {
-                       assert_eq!(node.node.get_and_clear_pending_events().len(), 0);
-                       assert_eq!(node.chan_monitor.added_monitors.lock().unwrap().len(), 0);
-               }
        }
 
        #[test]
@@ -3352,12 +3413,6 @@ mod tests {
                get_announce_close_broadcast_events(&nodes, 0, 1);
                assert_eq!(nodes[0].node.list_channels().len(), 0);
                assert_eq!(nodes[1].node.list_channels().len(), 0);
-
-               // Check that we processed all pending events
-               for node in nodes {
-                       assert_eq!(node.node.get_and_clear_pending_events().len(), 0);
-                       assert_eq!(node.chan_monitor.added_monitors.lock().unwrap().len(), 0);
-               }
        }
 
        #[test]
@@ -3542,11 +3597,164 @@ mod tests {
                while !headers.is_empty() {
                        nodes[0].node.block_disconnected(&headers.pop().unwrap());
                }
+               {
+                       let events = nodes[0].node.get_and_clear_pending_events();
+                       assert_eq!(events.len(), 1);
+                       match events[0] {
+                               Event::BroadcastChannelUpdate { msg: msgs::ChannelUpdate { contents: msgs::UnsignedChannelUpdate { flags, .. }, .. } } => {
+                                       assert_eq!(flags & 0b10, 0b10);
+                               },
+                               _ => panic!("Unexpected event"),
+                       }
+               }
                let channel_state = nodes[0].node.channel_state.lock().unwrap();
                assert_eq!(channel_state.by_id.len(), 0);
                assert_eq!(channel_state.short_to_id.len(), 0);
        }
 
+       fn reconnect_nodes(node_a: &Node, node_b: &Node, pre_all_htlcs: bool, pending_htlc_claims: (usize, usize), pending_htlc_fails: (usize, usize)) {
+               let reestablish_1 = node_a.node.peer_connected(&node_b.node.get_our_node_id());
+               let reestablish_2 = node_b.node.peer_connected(&node_a.node.get_our_node_id());
+
+               let mut resp_1 = Vec::new();
+               for msg in reestablish_1 {
+                       resp_1.push(node_b.node.handle_channel_reestablish(&node_a.node.get_our_node_id(), &msg).unwrap());
+               }
+               {
+                       let mut added_monitors = node_b.chan_monitor.added_monitors.lock().unwrap();
+                       if pending_htlc_claims.0 != 0 || pending_htlc_fails.0 != 0 {
+                               assert_eq!(added_monitors.len(), 1);
+                       } else {
+                               assert!(added_monitors.is_empty());
+                       }
+                       added_monitors.clear();
+               }
+
+               let mut resp_2 = Vec::new();
+               for msg in reestablish_2 {
+                       resp_2.push(node_a.node.handle_channel_reestablish(&node_b.node.get_our_node_id(), &msg).unwrap());
+               }
+               {
+                       let mut added_monitors = node_a.chan_monitor.added_monitors.lock().unwrap();
+                       if pending_htlc_claims.1 != 0 || pending_htlc_fails.1 != 0 {
+                               assert_eq!(added_monitors.len(), 1);
+                       } else {
+                               assert!(added_monitors.is_empty());
+                       }
+                       added_monitors.clear();
+               }
+
+               // We dont yet support both needing updates, as that would require a different commitment dance:
+               assert!((pending_htlc_claims.0 == 0 && pending_htlc_fails.0 == 0) || (pending_htlc_claims.1 == 0 && pending_htlc_fails.1 == 0));
+
+               for chan_msgs in resp_1.drain(..) {
+                       if pre_all_htlcs {
+                               let _announcement_sigs_opt = node_a.node.handle_funding_locked(&node_b.node.get_our_node_id(), &chan_msgs.0.unwrap()).unwrap();
+                               //TODO: Test announcement_sigs re-sending when we've implemented it
+                       } else {
+                               assert!(chan_msgs.0.is_none());
+                       }
+                       assert!(chan_msgs.1.is_none());
+                       if pending_htlc_claims.0 != 0 || pending_htlc_fails.0 != 0 {
+                               let commitment_update = chan_msgs.2.unwrap();
+                               assert!(commitment_update.update_add_htlcs.is_empty()); // We can't relay while disconnected
+                               assert_eq!(commitment_update.update_fulfill_htlcs.len(), pending_htlc_claims.0);
+                               assert_eq!(commitment_update.update_fail_htlcs.len(), pending_htlc_fails.0);
+                               assert!(commitment_update.update_fail_malformed_htlcs.is_empty());
+                               for update_fulfill in commitment_update.update_fulfill_htlcs {
+                                       node_a.node.handle_update_fulfill_htlc(&node_b.node.get_our_node_id(), &update_fulfill).unwrap();
+                               }
+                               for update_fail in commitment_update.update_fail_htlcs {
+                                       node_a.node.handle_update_fail_htlc(&node_b.node.get_our_node_id(), &update_fail).unwrap();
+                               }
+
+                               commitment_signed_dance!(node_a, node_b, commitment_update.commitment_signed, false);
+                       } else {
+                               assert!(chan_msgs.2.is_none());
+                       }
+               }
+
+               for chan_msgs in resp_2.drain(..) {
+                       if pre_all_htlcs {
+                               let _announcement_sigs_opt = node_b.node.handle_funding_locked(&node_a.node.get_our_node_id(), &chan_msgs.0.unwrap()).unwrap();
+                               //TODO: Test announcement_sigs re-sending when we've implemented it
+                       } else {
+                               assert!(chan_msgs.0.is_none());
+                       }
+                       assert!(chan_msgs.1.is_none());
+                       if pending_htlc_claims.1 != 0 || pending_htlc_fails.1 != 0 {
+                               let commitment_update = chan_msgs.2.unwrap();
+                               assert!(commitment_update.update_add_htlcs.is_empty()); // We can't relay while disconnected
+                               assert_eq!(commitment_update.update_fulfill_htlcs.len(), pending_htlc_claims.0);
+                               assert_eq!(commitment_update.update_fail_htlcs.len(), pending_htlc_fails.0);
+                               assert!(commitment_update.update_fail_malformed_htlcs.is_empty());
+                               for update_fulfill in commitment_update.update_fulfill_htlcs {
+                                       node_b.node.handle_update_fulfill_htlc(&node_a.node.get_our_node_id(), &update_fulfill).unwrap();
+                               }
+                               for update_fail in commitment_update.update_fail_htlcs {
+                                       node_b.node.handle_update_fail_htlc(&node_a.node.get_our_node_id(), &update_fail).unwrap();
+                               }
+
+                               commitment_signed_dance!(node_b, node_a, commitment_update.commitment_signed, false);
+                       } else {
+                               assert!(chan_msgs.2.is_none());
+                       }
+               }
+       }
+
+       #[test]
+       fn test_simple_peer_disconnect() {
+               // Test that we can reconnect when there are no lost messages
+               let nodes = create_network(3);
+               create_announced_chan_between_nodes(&nodes, 0, 1);
+               create_announced_chan_between_nodes(&nodes, 1, 2);
+
+               nodes[0].node.peer_disconnected(&nodes[1].node.get_our_node_id(), false);
+               nodes[1].node.peer_disconnected(&nodes[0].node.get_our_node_id(), false);
+               reconnect_nodes(&nodes[0], &nodes[1], true, (0, 0), (0, 0));
+
+               let payment_preimage_1 = route_payment(&nodes[0], &vec!(&nodes[1], &nodes[2])[..], 1000000).0;
+               let payment_hash_2 = route_payment(&nodes[0], &vec!(&nodes[1], &nodes[2])[..], 1000000).1;
+               fail_payment(&nodes[0], &vec!(&nodes[1], &nodes[2]), payment_hash_2);
+               claim_payment(&nodes[0], &vec!(&nodes[1], &nodes[2]), payment_preimage_1);
+
+               nodes[0].node.peer_disconnected(&nodes[1].node.get_our_node_id(), false);
+               nodes[1].node.peer_disconnected(&nodes[0].node.get_our_node_id(), false);
+               reconnect_nodes(&nodes[0], &nodes[1], false, (0, 0), (0, 0));
+
+               let payment_preimage_3 = route_payment(&nodes[0], &vec!(&nodes[1], &nodes[2])[..], 1000000).0;
+               let payment_preimage_4 = route_payment(&nodes[0], &vec!(&nodes[1], &nodes[2])[..], 1000000).0;
+               let payment_hash_5 = route_payment(&nodes[0], &vec!(&nodes[1], &nodes[2])[..], 1000000).1;
+               let payment_hash_6 = route_payment(&nodes[0], &vec!(&nodes[1], &nodes[2])[..], 1000000).1;
+
+               nodes[0].node.peer_disconnected(&nodes[1].node.get_our_node_id(), false);
+               nodes[1].node.peer_disconnected(&nodes[0].node.get_our_node_id(), false);
+
+               claim_payment_along_route(&nodes[0], &vec!(&nodes[1], &nodes[2]), true, payment_preimage_3);
+               fail_payment_along_route(&nodes[0], &[&nodes[1], &nodes[2]], true, payment_hash_5);
+
+               reconnect_nodes(&nodes[0], &nodes[1], false, (1, 0), (1, 0));
+               {
+                       let events = nodes[0].node.get_and_clear_pending_events();
+                       assert_eq!(events.len(), 2);
+                       match events[0] {
+                               Event::PaymentSent { payment_preimage } => {
+                                       assert_eq!(payment_preimage, payment_preimage_3);
+                               },
+                               _ => panic!("Unexpected event"),
+                       }
+                       match events[1] {
+                               Event::PaymentFailed { payment_hash } => {
+                                       assert_eq!(payment_hash, payment_hash_5);
+                               },
+                               _ => panic!("Unexpected event"),
+                       }
+               }
+
+               claim_payment(&nodes[0], &vec!(&nodes[1], &nodes[2]), payment_preimage_4);
+               fail_payment(&nodes[0], &vec!(&nodes[1], &nodes[2]), payment_hash_6);
+       }
+
        #[test]
        fn test_invalid_channel_announcement() {
                //Test BOLT 7 channel_announcement msg requirement for final node, gather data to build customed channel_announcement msgs