X-Git-Url: http://git.bitcoin.ninja/index.cgi?a=blobdiff_plain;f=src%2Fln%2Fchannelmanager.rs;h=5ee2f6725566fcb818eaf306f252ca62c182188a;hb=224fb05cc53526e7b861e746183451d3d74ad549;hp=bd663c9241235e7dee611f709e71541b9cf3ceac;hpb=e397cb99601e9d2849bbc3aad2b0df8bc8b7f522;p=rust-lightning diff --git a/src/ln/channelmanager.rs b/src/ln/channelmanager.rs index bd663c92..5ee2f672 100644 --- a/src/ln/channelmanager.rs +++ b/src/ln/channelmanager.rs @@ -23,14 +23,15 @@ use secp256k1; use chain::chaininterface::{BroadcasterInterface,ChainListener,ChainWatchInterface,FeeEstimator}; use chain::transaction::OutPoint; use ln::channel::{Channel, ChannelError}; -use ln::channelmonitor::{ChannelMonitorUpdateErr, ManyChannelMonitor, CLTV_CLAIM_BUFFER, HTLC_FAIL_TIMEOUT_BLOCKS}; +use ln::channelmonitor::{ChannelMonitor, ChannelMonitorUpdateErr, ManyChannelMonitor, CLTV_CLAIM_BUFFER, HTLC_FAIL_TIMEOUT_BLOCKS}; use ln::router::{Route,RouteHop}; use ln::msgs; -use ln::msgs::{ChannelMessageHandler, HandleError, RAACommitmentOrder}; +use ln::msgs::{ChannelMessageHandler, DecodeError, HandleError}; use chain::keysinterface::KeysInterface; +use util::config::UserConfig; use util::{byte_utils, events, internal_traits, rng}; use util::sha2::Sha256; -use util::ser::{Readable, Writeable}; +use util::ser::{Readable, ReadableArgs, Writeable, Writer}; use util::chacha20poly1305rfc::ChaCha20; use util::logger::Logger; use util::errors::APIError; @@ -41,11 +42,10 @@ use crypto::hmac::Hmac; use crypto::digest::Digest; use crypto::symmetriccipher::SynchronousStreamCipher; -use std::{ptr, mem}; -use std::collections::HashMap; -use std::collections::hash_map; +use std::{cmp, ptr, mem}; +use std::collections::{HashMap, hash_map, HashSet}; use std::io::Cursor; -use std::sync::{Mutex,MutexGuard,Arc}; +use std::sync::{Arc, Mutex, MutexGuard, RwLock}; use std::sync::atomic::{AtomicUsize, Ordering}; use std::time::{Instant,Duration}; @@ -65,13 +65,12 @@ mod channel_held_info { use ln::msgs; use ln::router::Route; use secp256k1::key::SecretKey; - use secp256k1::ecdh::SharedSecret; /// Stores the info we will need to send when we want to forward an HTLC onwards #[derive(Clone)] // See Channel::revoke_and_ack for why, tl;dr: Rust bug pub struct PendingForwardHTLCInfo { pub(super) onion_packet: Option, - pub(super) incoming_shared_secret: SharedSecret, + pub(super) incoming_shared_secret: [u8; 32], pub(super) payment_hash: [u8; 32], pub(super) short_channel_id: u64, pub(super) amt_to_forward: u64, @@ -96,7 +95,7 @@ mod channel_held_info { pub struct HTLCPreviousHopData { pub(super) short_channel_id: u64, pub(super) htlc_id: u64, - pub(super) incoming_packet_shared_secret: SharedSecret, + pub(super) incoming_packet_shared_secret: [u8; 32], } /// Tracks the inbound corresponding to an outbound HTLC @@ -244,6 +243,18 @@ struct HTLCForwardInfo { forward_info: PendingForwardHTLCInfo, } +/// For events which result in both a RevokeAndACK and a CommitmentUpdate, by default they should +/// be sent in the order they appear in the return value, however sometimes the order needs to be +/// variable at runtime (eg Channel::channel_reestablish needs to re-send messages in the order +/// they were originally sent). In those cases, this enum is also returned. +#[derive(Clone, PartialEq)] +pub(super) enum RAACommitmentOrder { + /// Send the CommitmentUpdate messages first + CommitmentFirst, + /// Send the RevokeAndACK message first + RevokeAndACKFirst, +} + struct ChannelHolder { by_id: HashMap<[u8; 32], Channel>, short_to_id: HashMap, @@ -290,22 +301,45 @@ const ERR: () = "You need at least 32 bit pointers (well, usize, but we'll assum /// /// Implements ChannelMessageHandler, handling the multi-channel parts and passing things through /// to individual Channels. +/// +/// Implements Writeable to write out all channel state to disk. Implies peer_disconnected() for +/// all peers during write/read (though does not modify this instance, only the instance being +/// serialized). This will result in any channels which have not yet exchanged funding_created (ie +/// called funding_transaction_generated for outbound channels). +/// +/// Note that you can be a bit lazier about writing out ChannelManager than you can be with +/// ChannelMonitors. With ChannelMonitors you MUST write each monitor update out to disk before +/// returning from ManyChannelMonitor::add_update_monitor, with ChannelManagers, writing updates +/// happens out-of-band (and will prevent any other ChannelManager operations from occurring during +/// the serialization process). If the deserialized version is out-of-date compared to the +/// ChannelMonitors passed by reference to read(), those channels will be force-closed based on the +/// ChannelMonitor state and no funds will be lost (mod on-chain transaction fees). +/// +/// Note that the deserializer is only implemented for (Sha256dHash, ChannelManager), which +/// tells you the last block hash which was block_connect()ed. You MUST rescan any blocks along +/// the "reorg path" (ie call block_disconnected() until you get to a common block and then call +/// block_connected() to step towards your best block) upon deserialization before using the +/// object! pub struct ChannelManager { + default_configuration: UserConfig, genesis_hash: Sha256dHash, fee_estimator: Arc, monitor: Arc, chain_monitor: Arc, tx_broadcaster: Arc, - announce_channels_publicly: bool, - fee_proportional_millionths: u32, latest_block_height: AtomicUsize, + last_block_hash: Mutex, secp_ctx: Secp256k1, channel_state: Mutex, our_network_key: SecretKey, pending_events: Mutex>, + /// 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, @@ -377,23 +411,22 @@ impl ChannelManager { /// This is the main "logic hub" for all channel-related actions, and implements /// ChannelMessageHandler. /// - /// fee_proportional_millionths is an optional fee to charge any payments routed through us. /// Non-proportional fees are fixed according to our risk using the provided fee estimator. /// /// panics if channel_value_satoshis is >= `MAX_FUNDING_SATOSHIS`! - pub fn new(fee_proportional_millionths: u32, announce_channels_publicly: bool, network: Network, feeest: Arc, monitor: Arc, chain_monitor: Arc, tx_broadcaster: Arc, logger: Arc, keys_manager: Arc) -> Result, secp256k1::Error> { + pub fn new(network: Network, feeest: Arc, monitor: Arc, chain_monitor: Arc, tx_broadcaster: Arc, logger: Arc,keys_manager: Arc, config: UserConfig) -> Result, secp256k1::Error> { let secp_ctx = Secp256k1::new(); let res = Arc::new(ChannelManager { + default_configuration: config.clone(), genesis_hash: genesis_block(network).header.bitcoin_hash(), fee_estimator: feeest.clone(), monitor: monitor.clone(), chain_monitor, tx_broadcaster, - announce_channels_publicly, - fee_proportional_millionths, - latest_block_height: AtomicUsize::new(0), //TODO: Get an init value (generally need to replay recent chain on chain_monitor registration) + latest_block_height: AtomicUsize::new(0), //TODO: Get an init value + last_block_hash: Mutex::new(Default::default()), secp_ctx, channel_state: Mutex::new(ChannelHolder{ @@ -407,6 +440,7 @@ impl ChannelManager { our_network_key: keys_manager.get_node_secret(), pending_events: Mutex::new(Vec::new()), + total_consistency_lock: RwLock::new(()), keys_manager, @@ -427,10 +461,17 @@ impl ChannelManager { /// If successful, will generate a SendOpenChannel message event, so you should probably poll /// PeerManager::process_events afterwards. /// - /// Raises APIError::APIMisuseError when channel_value_satoshis > 2**24 or push_msat being greater than channel_value_satoshis * 1k + /// Raises APIError::APIMisuseError when channel_value_satoshis > 2**24 or push_msat is + /// greater than channel_value_satoshis * 1k or channel_value_satoshis is < 1000. pub fn create_channel(&self, their_network_key: PublicKey, channel_value_satoshis: u64, push_msat: u64, user_id: u64) -> Result<(), APIError> { - let channel = Channel::new_outbound(&*self.fee_estimator, &self.keys_manager, their_network_key, channel_value_satoshis, push_msat, self.announce_channels_publicly, user_id, Arc::clone(&self.logger))?; + if channel_value_satoshis < 1000 { + return Err(APIError::APIMisuseError { err: "channel_value must be at least 1000 satoshis" }); + } + + let channel = Channel::new_outbound(&*self.fee_estimator, &self.keys_manager, their_network_key, channel_value_satoshis, push_msat, user_id, Arc::clone(&self.logger), &self.default_configuration)?; let res = channel.get_open_channel(self.genesis_hash.clone(), &*self.fee_estimator); + + let _ = self.total_consistency_lock.read().unwrap(); let mut channel_state = self.channel_state.lock().unwrap(); match channel_state.by_id.entry(channel.channel_id()) { hash_map::Entry::Occupied(_) => { @@ -494,6 +535,8 @@ impl ChannelManager { /// /// May generate a SendShutdown message event on success, which should be relayed. pub fn close_channel(&self, channel_id: &[u8; 32]) -> Result<(), APIError> { + let _ = self.total_consistency_lock.read().unwrap(); + let (mut failed_htlcs, chan_option) = { let mut channel_state_lock = self.channel_state.lock().unwrap(); let channel_state = channel_state_lock.borrow_parts(); @@ -556,6 +599,8 @@ impl ChannelManager { /// Force closes a channel, immediately broadcasting the latest local commitment transaction to /// the chain and rejecting new HTLCs on the given channel. pub fn force_close_channel(&self, channel_id: &[u8; 32]) { + let _ = self.total_consistency_lock.read().unwrap(); + let mut chan = { let mut channel_state_lock = self.channel_state.lock().unwrap(); let channel_state = channel_state_lock.borrow_parts(); @@ -613,7 +658,8 @@ impl ChannelManager { } #[inline] - fn gen_rho_mu_from_shared_secret(shared_secret: &SharedSecret) -> ([u8; 32], [u8; 32]) { + fn gen_rho_mu_from_shared_secret(shared_secret: &[u8]) -> ([u8; 32], [u8; 32]) { + assert_eq!(shared_secret.len(), 32); ({ let mut hmac = Hmac::new(Sha256::new(), &[0x72, 0x68, 0x6f]); // rho hmac.input(&shared_secret[..]); @@ -631,7 +677,8 @@ impl ChannelManager { } #[inline] - fn gen_um_from_shared_secret(shared_secret: &SharedSecret) -> [u8; 32] { + fn gen_um_from_shared_secret(shared_secret: &[u8]) -> [u8; 32] { + assert_eq!(shared_secret.len(), 32); let mut hmac = Hmac::new(Sha256::new(), &[0x75, 0x6d]); // um hmac.input(&shared_secret[..]); let mut res = [0; 32]; @@ -640,7 +687,8 @@ impl ChannelManager { } #[inline] - fn gen_ammag_from_shared_secret(shared_secret: &SharedSecret) -> [u8; 32] { + fn gen_ammag_from_shared_secret(shared_secret: &[u8]) -> [u8; 32] { + assert_eq!(shared_secret.len(), 32); let mut hmac = Hmac::new(Sha256::new(), &[0x61, 0x6d, 0x6d, 0x61, 0x67]); // ammag hmac.input(&shared_secret[..]); let mut res = [0; 32]; @@ -679,7 +727,7 @@ impl ChannelManager { let mut res = Vec::with_capacity(route.hops.len()); Self::construct_onion_keys_callback(secp_ctx, route, session_priv, |shared_secret, _blinding_factor, ephemeral_pubkey, _| { - let (rho, mu) = ChannelManager::gen_rho_mu_from_shared_secret(&shared_secret); + let (rho, mu) = ChannelManager::gen_rho_mu_from_shared_secret(&shared_secret[..]); res.push(OnionKeys { #[cfg(test)] @@ -803,7 +851,7 @@ impl ChannelManager { /// Encrypts a failure packet. raw_packet can either be a /// msgs::DecodedOnionErrorPacket.encode() result or a msgs::OnionErrorPacket.data element. - fn encrypt_failure_packet(shared_secret: &SharedSecret, raw_packet: &[u8]) -> msgs::OnionErrorPacket { + fn encrypt_failure_packet(shared_secret: &[u8], raw_packet: &[u8]) -> msgs::OnionErrorPacket { let ammag = ChannelManager::gen_ammag_from_shared_secret(&shared_secret); let mut packet_crypted = Vec::with_capacity(raw_packet.len()); @@ -815,7 +863,8 @@ impl ChannelManager { } } - fn build_failure_packet(shared_secret: &SharedSecret, failure_type: u16, failure_data: &[u8]) -> msgs::DecodedOnionErrorPacket { + fn build_failure_packet(shared_secret: &[u8], failure_type: u16, failure_data: &[u8]) -> msgs::DecodedOnionErrorPacket { + assert_eq!(shared_secret.len(), 32); assert!(failure_data.len() <= 256 - 2); let um = ChannelManager::gen_um_from_shared_secret(&shared_secret); @@ -846,7 +895,7 @@ impl ChannelManager { } #[inline] - fn build_first_hop_failure_packet(shared_secret: &SharedSecret, failure_type: u16, failure_data: &[u8]) -> msgs::OnionErrorPacket { + fn build_first_hop_failure_packet(shared_secret: &[u8], failure_type: u16, failure_data: &[u8]) -> msgs::OnionErrorPacket { let failure_packet = ChannelManager::build_failure_packet(shared_secret, failure_type, failure_data); ChannelManager::encrypt_failure_packet(shared_secret, &failure_packet.encode()[..]) } @@ -874,7 +923,11 @@ impl ChannelManager { })), self.channel_state.lock().unwrap()); } - let shared_secret = SharedSecret::new(&self.secp_ctx, &msg.onion_routing_packet.public_key.unwrap(), &self.our_network_key); + let shared_secret = { + let mut arr = [0; 32]; + arr.copy_from_slice(&SharedSecret::new(&self.secp_ctx, &msg.onion_routing_packet.public_key.unwrap(), &self.our_network_key)[..]); + arr + }; let (rho, mu) = ChannelManager::gen_rho_mu_from_shared_secret(&shared_secret); let mut channel_state = None; @@ -951,7 +1004,7 @@ impl ChannelManager { onion_packet: None, payment_hash: msg.payment_hash.clone(), short_channel_id: 0, - incoming_shared_secret: shared_secret.clone(), + incoming_shared_secret: shared_secret, amt_to_forward: next_hop_data.data.amt_to_forward, outgoing_cltv_value: next_hop_data.data.outgoing_cltv_value, }) @@ -965,7 +1018,7 @@ impl ChannelManager { let blinding_factor = { let mut sha = Sha256::new(); sha.input(&new_pubkey.serialize()[..]); - sha.input(&shared_secret[..]); + sha.input(&shared_secret); let mut res = [0u8; 32]; sha.result(&mut res); match SecretKey::from_slice(&self.secp_ctx, &res) { @@ -991,7 +1044,7 @@ impl ChannelManager { onion_packet: Some(outgoing_packet), payment_hash: msg.payment_hash.clone(), short_channel_id: next_hop_data.data.short_channel_id, - incoming_shared_secret: shared_secret.clone(), + incoming_shared_secret: shared_secret, amt_to_forward: next_hop_data.data.amt_to_forward, outgoing_cltv_value: next_hop_data.data.outgoing_cltv_value, }) @@ -1021,7 +1074,7 @@ impl ChannelManager { if *amt_to_forward < chan.get_their_htlc_minimum_msat() { // amount_below_minimum break Some(("HTLC amount was below the htlc_minimum_msat", 0x1000 | 11, Some(self.get_channel_update(chan).unwrap()))); } - let fee = amt_to_forward.checked_mul(self.fee_proportional_millionths as u64).and_then(|prop_fee| { (prop_fee / 1000000).checked_add(chan.get_our_fee_base_msat(&*self.fee_estimator) as u64) }); + let fee = amt_to_forward.checked_mul(chan.get_fee_proportional_millionths() as u64).and_then(|prop_fee| { (prop_fee / 1000000).checked_add(chan.get_our_fee_base_msat(&*self.fee_estimator) as u64) }); if fee.is_none() || msg.amount_msat < fee.unwrap() || (msg.amount_msat - fee.unwrap()) < *amt_to_forward { // fee_insufficient break Some(("Prior hop has deviated from specified fees parameters or origin node has obsolete ones", 0x1000 | 12, Some(self.get_channel_update(chan).unwrap()))); } @@ -1075,7 +1128,7 @@ impl ChannelManager { cltv_expiry_delta: CLTV_EXPIRY_DELTA, htlc_minimum_msat: chan.get_our_htlc_minimum_msat(), fee_base_msat: chan.get_our_fee_base_msat(&*self.fee_estimator), - fee_proportional_millionths: self.fee_proportional_millionths, + fee_proportional_millionths: chan.get_fee_proportional_millionths(), excess_data: Vec::new(), }; @@ -1128,6 +1181,7 @@ impl ChannelManager { let (onion_payloads, htlc_msat, htlc_cltv) = ChannelManager::build_onion_payloads(&route, cur_height)?; let onion_packet = ChannelManager::construct_onion_packet(onion_payloads, onion_keys, &payment_hash); + let _ = self.total_consistency_lock.read().unwrap(); let mut channel_state = self.channel_state.lock().unwrap(); let id = match channel_state.short_to_id.get(&route.hops.first().unwrap().short_channel_id) { @@ -1179,11 +1233,16 @@ impl ChannelManager { /// Call this upon creation of a funding transaction for the given channel. /// + /// Note that ALL inputs in the transaction pointed to by funding_txo MUST spend SegWit outputs + /// or your counterparty can steal your funds! + /// /// Panics if a funding transaction has already been provided for this channel. /// /// May panic if the funding_txo is duplicative with some other channel (note that this should /// be trivially prevented by using unique funding transaction keys per-channel). 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 channel_state = self.channel_state.lock().unwrap(); match channel_state.by_id.remove(temporary_channel_id) { @@ -1249,6 +1308,8 @@ impl ChannelManager { /// Should only really ever be called in response to an PendingHTLCsForwardable event. /// Will likely generate further events. pub fn process_pending_htlc_forwards(&self) { + let _ = self.total_consistency_lock.read().unwrap(); + let mut new_events = Vec::new(); let mut failed_forwards = Vec::new(); { @@ -1370,6 +1431,8 @@ impl ChannelManager { /// Indicates that the preimage for payment_hash is unknown or the received amount is incorrect after a PaymentReceived event. pub fn fail_htlc_backwards(&self, payment_hash: &[u8; 32], reason: PaymentFailReason) -> bool { + let _ = self.total_consistency_lock.read().unwrap(); + let mut channel_state = Some(self.channel_state.lock().unwrap()); let removed_source = channel_state.as_mut().unwrap().claimable_htlcs.remove(payment_hash); if let Some(mut sources) = removed_source { @@ -1465,6 +1528,8 @@ impl ChannelManager { let mut payment_hash = [0; 32]; sha.result(&mut payment_hash); + let _ = self.total_consistency_lock.read().unwrap(); + let mut channel_state = Some(self.channel_state.lock().unwrap()); let removed_source = channel_state.as_mut().unwrap().claimable_htlcs.remove(&payment_hash); if let Some(mut sources) = removed_source { @@ -1543,6 +1608,7 @@ impl ChannelManager { let mut close_results = Vec::new(); let mut htlc_forwards = Vec::new(); let mut htlc_failures = Vec::new(); + let _ = self.total_consistency_lock.read().unwrap(); { let mut channel_lock = self.channel_state.lock().unwrap(); @@ -1617,20 +1683,26 @@ impl ChannelManager { } } - fn internal_open_channel(&self, their_node_id: &PublicKey, msg: &msgs::OpenChannel) -> Result { + fn internal_open_channel(&self, their_node_id: &PublicKey, 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 mut channel_state = self.channel_state.lock().unwrap(); - if channel_state.by_id.contains_key(&msg.temporary_channel_id) { - return Err(MsgHandleErrInternal::send_err_msg_no_close("temporary_channel_id collision!", msg.temporary_channel_id.clone())); - } - let channel = Channel::new_from_req(&*self.fee_estimator, &self.keys_manager, their_node_id.clone(), msg, 0, false, self.announce_channels_publicly, Arc::clone(&self.logger)) + 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) .map_err(|e| MsgHandleErrInternal::from_chan_no_close(e, msg.temporary_channel_id))?; - let accept_msg = channel.get_accept_channel(); - channel_state.by_id.insert(channel.channel_id(), channel); - Ok(accept_msg) + let mut channel_state_lock = self.channel_state.lock().unwrap(); + let channel_state = channel_state_lock.borrow_parts(); + 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) => { + channel_state.pending_msg_events.push(events::MessageSendEvent::SendAcceptChannel { + node_id: their_node_id.clone(), + msg: channel.get_accept_channel(), + }); + entry.insert(channel); + } + } + Ok(()) } fn internal_accept_channel(&self, their_node_id: &PublicKey, msg: &msgs::AcceptChannel) -> Result<(), MsgHandleErrInternal> { @@ -1642,7 +1714,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)); } - chan.accept_channel(&msg) + chan.accept_channel(&msg, &self.default_configuration) .map_err(|e| MsgHandleErrInternal::from_chan_maybe_close(e, msg.temporary_channel_id))?; (chan.get_value_satoshis(), chan.get_funding_redeemscript().to_v0_p2wsh(), chan.get_user_id()) }, @@ -1660,7 +1732,7 @@ impl ChannelManager { Ok(()) } - fn internal_funding_created(&self, their_node_id: &PublicKey, msg: &msgs::FundingCreated) -> Result { + fn internal_funding_created(&self, their_node_id: &PublicKey, msg: &msgs::FundingCreated) -> Result<(), MsgHandleErrInternal> { let (chan, funding_msg, monitor_update) = { let mut channel_state = self.channel_state.lock().unwrap(); match channel_state.by_id.entry(msg.temporary_channel_id.clone()) { @@ -1686,16 +1758,21 @@ impl ChannelManager { if let Err(_e) = self.monitor.add_update_monitor(monitor_update.get_funding_txo().unwrap(), monitor_update) { unimplemented!(); } - let mut channel_state = self.channel_state.lock().unwrap(); + let mut channel_state_lock = self.channel_state.lock().unwrap(); + let channel_state = channel_state_lock.borrow_parts(); 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)) }, hash_map::Entry::Vacant(e) => { + channel_state.pending_msg_events.push(events::MessageSendEvent::SendFundingSigned { + node_id: their_node_id.clone(), + msg: funding_msg, + }); e.insert(chan); } } - Ok(funding_msg) + Ok(()) } fn internal_funding_signed(&self, their_node_id: &PublicKey, msg: &msgs::FundingSigned) -> Result<(), MsgHandleErrInternal> { @@ -1724,8 +1801,9 @@ impl ChannelManager { Ok(()) } - fn internal_funding_locked(&self, their_node_id: &PublicKey, msg: &msgs::FundingLocked) -> Result, MsgHandleErrInternal> { - let mut channel_state = self.channel_state.lock().unwrap(); + 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(); match channel_state.by_id.get_mut(&msg.channel_id) { Some(chan) => { if chan.get_their_node_id() != *their_node_id { @@ -1734,14 +1812,20 @@ impl ChannelManager { } chan.funding_locked(&msg) .map_err(|e| MsgHandleErrInternal::from_chan_maybe_close(e, msg.channel_id))?; - return Ok(self.get_announcement_sigs(chan)); + if let Some(announcement_sigs) = self.get_announcement_sigs(chan) { + channel_state.pending_msg_events.push(events::MessageSendEvent::SendAnnouncementSignatures { + node_id: their_node_id.clone(), + msg: announcement_sigs, + }); + } + Ok(()) }, - None => return Err(MsgHandleErrInternal::send_err_msg_no_close("Failed to find corresponding channel", msg.channel_id)) - }; + None => Err(MsgHandleErrInternal::send_err_msg_no_close("Failed to find corresponding channel", msg.channel_id)) + } } - fn internal_shutdown(&self, their_node_id: &PublicKey, msg: &msgs::Shutdown) -> Result<(Option, Option), MsgHandleErrInternal> { - let (mut res, chan_option) = { + 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(); @@ -1751,18 +1835,30 @@ 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)); } - let res = chan_entry.get_mut().shutdown(&*self.fee_estimator, &msg).map_err(|e| MsgHandleErrInternal::from_maybe_close(e))?; + let (shutdown, closing_signed, dropped_htlcs) = chan_entry.get_mut().shutdown(&*self.fee_estimator, &msg).map_err(|e| MsgHandleErrInternal::from_chan_maybe_close(e, msg.channel_id))?; + if let Some(msg) = shutdown { + channel_state.pending_msg_events.push(events::MessageSendEvent::SendShutdown { + node_id: their_node_id.clone(), + msg, + }); + } + if let Some(msg) = closing_signed { + channel_state.pending_msg_events.push(events::MessageSendEvent::SendClosingSigned { + node_id: their_node_id.clone(), + msg, + }); + } if chan_entry.get().is_shutdown() { if let Some(short_id) = chan_entry.get().get_short_channel_id() { channel_state.short_to_id.remove(&short_id); } - (res, Some(chan_entry.remove_entry().1)) - } else { (res, None) } + (dropped_htlcs, Some(chan_entry.remove_entry().1)) + } else { (dropped_htlcs, None) } }, hash_map::Entry::Vacant(_) => return Err(MsgHandleErrInternal::send_err_msg_no_close("Failed to find corresponding channel", msg.channel_id)) } }; - for htlc_source in res.2.drain(..) { + for htlc_source in dropped_htlcs.drain(..) { // unknown_next_peer...I dunno who that is anymore.... self.fail_htlc_backwards_internal(self.channel_state.lock().unwrap(), htlc_source.0, &htlc_source.1, HTLCFailReason::Reason { failure_code: 0x4000 | 10, data: Vec::new() }); } @@ -1774,11 +1870,11 @@ impl ChannelManager { }); } } - Ok((res.0, res.1)) + Ok(()) } - fn internal_closing_signed(&self, their_node_id: &PublicKey, msg: &msgs::ClosingSigned) -> Result, MsgHandleErrInternal> { - let (res, chan_option) = { + 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(); match channel_state.by_id.entry(msg.channel_id.clone()) { @@ -1787,8 +1883,14 @@ 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)); } - let res = chan_entry.get_mut().closing_signed(&*self.fee_estimator, &msg).map_err(|e| MsgHandleErrInternal::from_maybe_close(e))?; - if res.1.is_some() { + let (closing_signed, tx) = chan_entry.get_mut().closing_signed(&*self.fee_estimator, &msg).map_err(|e| MsgHandleErrInternal::from_maybe_close(e))?; + if let Some(msg) = closing_signed { + channel_state.pending_msg_events.push(events::MessageSendEvent::SendClosingSigned { + node_id: their_node_id.clone(), + msg, + }); + } + if tx.is_some() { // We're done with this channel, we've got a signed closing transaction and // will send the closing_signed back to the remote peer upon return. This // also implies there are no pending HTLCs left on the channel, so we can @@ -1797,13 +1899,13 @@ impl ChannelManager { if let Some(short_id) = chan_entry.get().get_short_channel_id() { channel_state.short_to_id.remove(&short_id); } - (res, Some(chan_entry.remove_entry().1)) - } else { (res, None) } + (tx, Some(chan_entry.remove_entry().1)) + } else { (tx, None) } }, hash_map::Entry::Vacant(_) => return Err(MsgHandleErrInternal::send_err_msg_no_close("Failed to find corresponding channel", msg.channel_id)) } }; - if let Some(broadcast_tx) = res.1 { + if let Some(broadcast_tx) = tx { self.tx_broadcaster.broadcast_transaction(&broadcast_tx); } if let Some(chan) = chan_option { @@ -1814,7 +1916,7 @@ impl ChannelManager { }); } } - Ok(res.0) + Ok(()) } fn internal_update_add_htlc(&self, their_node_id: &PublicKey, msg: &msgs::UpdateAddHTLC) -> Result<(), MsgHandleErrInternal> { @@ -1827,7 +1929,7 @@ impl ChannelManager { //encrypted with the same key. Its not immediately obvious how to usefully exploit that, //but we should prevent it anyway. - let (pending_forward_info, mut channel_state_lock) = self.decode_update_add_htlc_onion(msg); + let (mut pending_forward_info, mut channel_state_lock) = self.decode_update_add_htlc_onion(msg); let channel_state = channel_state_lock.borrow_parts(); match channel_state.by_id.get_mut(&msg.channel_id) { @@ -1837,7 +1939,16 @@ 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.is_usable() { - return Err(MsgHandleErrInternal::from_no_close(HandleError{err: "Channel not yet available for receiving HTLCs", action: Some(msgs::ErrorAction::IgnoreError)})); + // If the update_add is completely bogus, the call will Err and we will close, + // but if we've sent a shutdown and they haven't acknowledged it yet, we just + // want to reject the new HTLC and fail it backwards instead of forwarding. + if let PendingHTLCStatus::Forward(PendingForwardHTLCInfo { incoming_shared_secret, .. }) = pending_forward_info { + pending_forward_info = PendingHTLCStatus::Fail(HTLCFailureMsg::Relay(msgs::UpdateFailHTLC { + channel_id: msg.channel_id, + htlc_id: msg.htlc_id, + reason: ChannelManager::build_first_hop_failure_packet(&incoming_shared_secret, 0x1000|20, &self.get_channel_update(chan).unwrap().encode_with_len()[..]), + })); + } } chan.update_add_htlc(&msg, pending_forward_info).map_err(|e| MsgHandleErrInternal::from_maybe_close(e)) }, @@ -1890,7 +2001,7 @@ impl ChannelManager { let amt_to_forward = htlc_msat - route_hop.fee_msat; htlc_msat = amt_to_forward; - let ammag = ChannelManager::gen_ammag_from_shared_secret(&shared_secret); + let ammag = ChannelManager::gen_ammag_from_shared_secret(&shared_secret[..]); let mut decryption_tmp = Vec::with_capacity(packet_decrypted.len()); decryption_tmp.resize(packet_decrypted.len(), 0); @@ -1901,7 +2012,7 @@ impl ChannelManager { let is_from_final_node = route.hops.last().unwrap().pubkey == route_hop.pubkey; if let Ok(err_packet) = msgs::DecodedOnionErrorPacket::read(&mut Cursor::new(&packet_decrypted)) { - let um = ChannelManager::gen_um_from_shared_secret(&shared_secret); + let um = ChannelManager::gen_um_from_shared_secret(&shared_secret[..]); let mut hmac = Hmac::new(Sha256::new(), &um); hmac.input(&err_packet.encode()[32..]); let mut calc_tag = [0u8; 32]; @@ -2094,25 +2205,46 @@ impl ChannelManager { } } - fn internal_commitment_signed(&self, their_node_id: &PublicKey, msg: &msgs::CommitmentSigned) -> Result<(msgs::RevokeAndACK, Option), MsgHandleErrInternal> { - let (revoke_and_ack, commitment_signed) = { - 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 { - //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, chan_monitor) = chan.commitment_signed(&msg).map_err(|e| MsgHandleErrInternal::from_maybe_close(e))?; - if let Err(_e) = self.monitor.add_update_monitor(chan_monitor.get_funding_txo().unwrap(), chan_monitor) { - unimplemented!(); - } - (revoke_and_ack, commitment_signed) - }, - None => return Err(MsgHandleErrInternal::send_err_msg_no_close("Failed to find corresponding channel", msg.channel_id)) - } - }; - Ok((revoke_and_ack, commitment_signed)) + 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(); + match channel_state.by_id.get_mut(&msg.channel_id) { + Some(chan) => { + if chan.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) = chan.commitment_signed(&msg, &*self.fee_estimator).map_err(|e| MsgHandleErrInternal::from_maybe_close(e))?; + if let Err(_e) = self.monitor.add_update_monitor(chan_monitor.get_funding_txo().unwrap(), chan_monitor) { + unimplemented!(); + } + channel_state.pending_msg_events.push(events::MessageSendEvent::SendRevokeAndACK { + node_id: their_node_id.clone(), + msg: revoke_and_ack, + }); + if let Some(msg) = commitment_signed { + channel_state.pending_msg_events.push(events::MessageSendEvent::UpdateHTLCs { + node_id: their_node_id.clone(), + updates: msgs::CommitmentUpdate { + update_add_htlcs: Vec::new(), + update_fulfill_htlcs: Vec::new(), + update_fail_htlcs: Vec::new(), + update_fail_malformed_htlcs: Vec::new(), + update_fee: None, + commitment_signed: msg, + }, + }); + } + if let Some(msg) = closing_signed { + channel_state.pending_msg_events.push(events::MessageSendEvent::SendClosingSigned { + node_id: their_node_id.clone(), + msg, + }); + } + Ok(()) + }, + None => Err(MsgHandleErrInternal::send_err_msg_no_close("Failed to find corresponding channel", msg.channel_id)) + } } #[inline] @@ -2148,20 +2280,33 @@ impl ChannelManager { } } - fn internal_revoke_and_ack(&self, their_node_id: &PublicKey, msg: &msgs::RevokeAndACK) -> Result, MsgHandleErrInternal> { - let ((res, pending_forwards, mut pending_failures), short_channel_id) = { - let mut channel_state = self.channel_state.lock().unwrap(); + 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(); match channel_state.by_id.get_mut(&msg.channel_id) { Some(chan) => { if chan.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 (res, pending_forwards, pending_failures, chan_monitor) = chan.revoke_and_ack(&msg).map_err(|e| MsgHandleErrInternal::from_maybe_close(e))?; + let (commitment_update, pending_forwards, pending_failures, closing_signed, chan_monitor) = chan.revoke_and_ack(&msg, &*self.fee_estimator).map_err(|e| MsgHandleErrInternal::from_maybe_close(e))?; if let Err(_e) = self.monitor.add_update_monitor(chan_monitor.get_funding_txo().unwrap(), chan_monitor) { unimplemented!(); } - ((res, pending_forwards, pending_failures), chan.get_short_channel_id().expect("RAA should only work on a short-id-available channel")) + if let Some(updates) = commitment_update { + channel_state.pending_msg_events.push(events::MessageSendEvent::UpdateHTLCs { + node_id: their_node_id.clone(), + updates, + }); + } + if let Some(msg) = closing_signed { + channel_state.pending_msg_events.push(events::MessageSendEvent::SendClosingSigned { + node_id: their_node_id.clone(), + msg, + }); + } + (pending_forwards, pending_failures, chan.get_short_channel_id().expect("RAA should only work on a short-id-available channel")) }, None => return Err(MsgHandleErrInternal::send_err_msg_no_close("Failed to find corresponding channel", msg.channel_id)) } @@ -2171,7 +2316,7 @@ impl ChannelManager { } self.forward_htlcs(&mut [(short_channel_id, pending_forwards)]); - Ok(res) + Ok(()) } fn internal_update_fee(&self, their_node_id: &PublicKey, msg: &msgs::UpdateFee) -> Result<(), MsgHandleErrInternal> { @@ -2229,28 +2374,64 @@ impl ChannelManager { Ok(()) } - fn internal_channel_reestablish(&self, their_node_id: &PublicKey, msg: &msgs::ChannelReestablish) -> Result<(Option, Option, Option, RAACommitmentOrder), MsgHandleErrInternal> { - let res = { - 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)); + 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(); + + 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, order, shutdown) = chan.channel_reestablish(msg) + .map_err(|e| MsgHandleErrInternal::from_chan_maybe_close(e, msg.channel_id))?; + if let Some(monitor) = channel_monitor { + if let Err(_e) = self.monitor.add_update_monitor(monitor.get_funding_txo().unwrap(), monitor) { + unimplemented!(); } - let (funding_locked, revoke_and_ack, commitment_update, channel_monitor, order) = chan.channel_reestablish(msg) - .map_err(|e| MsgHandleErrInternal::from_chan_maybe_close(e, msg.channel_id))?; - if let Some(monitor) = channel_monitor { - if let Err(_e) = self.monitor.add_update_monitor(monitor.get_funding_txo().unwrap(), monitor) { - unimplemented!(); - } + } + if let Some(msg) = funding_locked { + channel_state.pending_msg_events.push(events::MessageSendEvent::SendFundingLocked { + node_id: their_node_id.clone(), + msg + }); + } + macro_rules! send_raa { () => { + if let Some(msg) = revoke_and_ack { + channel_state.pending_msg_events.push(events::MessageSendEvent::SendRevokeAndACK { + node_id: their_node_id.clone(), + msg + }); } - Ok((funding_locked, revoke_and_ack, commitment_update, order)) - }, - None => return Err(MsgHandleErrInternal::send_err_msg_no_close("Failed to find corresponding channel", msg.channel_id)) - } - }; - - res + } } + macro_rules! send_cu { () => { + if let Some(updates) = commitment_update { + channel_state.pending_msg_events.push(events::MessageSendEvent::UpdateHTLCs { + node_id: their_node_id.clone(), + updates + }); + } + } } + match order { + RAACommitmentOrder::RevokeAndACKFirst => { + send_raa!(); + send_cu!(); + }, + RAACommitmentOrder::CommitmentFirst => { + send_cu!(); + send_raa!(); + }, + } + if let Some(msg) = shutdown { + channel_state.pending_msg_events.push(events::MessageSendEvent::SendShutdown { + node_id: their_node_id.clone(), + msg, + }); + } + Ok(()) + }, + None => Err(MsgHandleErrInternal::send_err_msg_no_close("Failed to find corresponding channel", msg.channel_id)) + } } /// Begin Update fee process. Allowed only on an outbound channel. @@ -2259,6 +2440,7 @@ impl ChannelManager { /// Note: This API is likely to change! #[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 channel_state = channel_state_lock.borrow_parts(); @@ -2316,6 +2498,7 @@ impl events::EventsProvider for ChannelManager { impl ChainListener for ChannelManager { fn block_connected(&self, header: &BlockHeader, height: u32, txn_matched: &[&Transaction], indexes_of_txn_matched: &[u32]) { + let _ = self.total_consistency_lock.read().unwrap(); let mut failed_channels = Vec::new(); { let mut channel_lock = self.channel_state.lock().unwrap(); @@ -2325,12 +2508,16 @@ impl ChainListener for ChannelManager { 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 { - let announcement_sigs = self.get_announcement_sigs(channel); pending_msg_events.push(events::MessageSendEvent::SendFundingLocked { node_id: channel.get_their_node_id(), msg: funding_locked, - announcement_sigs: announcement_sigs }); + 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()); } else if let Err(e) = chan_res { pending_msg_events.push(events::MessageSendEvent::HandleError { @@ -2385,10 +2572,12 @@ impl ChainListener for ChannelManager { self.finish_force_close_channel(failure); } self.latest_block_height.store(height as usize, Ordering::Release); + *self.last_block_hash.try_lock().expect("block_(dis)connected must not be called in parallel") = header.bitcoin_hash(); } /// We force-close the channel without letting our counterparty participate in the shutdown fn block_disconnected(&self, header: &BlockHeader) { + let _ = self.total_consistency_lock.read().unwrap(); let mut failed_channels = Vec::new(); { let mut channel_lock = self.channel_state.lock().unwrap(); @@ -2416,6 +2605,7 @@ impl ChainListener for ChannelManager { self.finish_force_close_channel(failure); } self.latest_block_height.fetch_sub(1, Ordering::AcqRel); + *self.last_block_hash.try_lock().expect("block_(dis)connected must not be called in parallel") = header.bitcoin_hash(); } } @@ -2453,71 +2643,88 @@ macro_rules! handle_error { 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 { + fn handle_open_channel(&self, their_node_id: &PublicKey, msg: &msgs::OpenChannel) -> Result<(), HandleError> { + let _ = self.total_consistency_lock.read().unwrap(); handle_error!(self, self.internal_open_channel(their_node_id, msg), their_node_id) } fn handle_accept_channel(&self, their_node_id: &PublicKey, msg: &msgs::AcceptChannel) -> Result<(), HandleError> { + let _ = self.total_consistency_lock.read().unwrap(); handle_error!(self, self.internal_accept_channel(their_node_id, msg), their_node_id) } - fn handle_funding_created(&self, their_node_id: &PublicKey, msg: &msgs::FundingCreated) -> Result { + fn handle_funding_created(&self, their_node_id: &PublicKey, msg: &msgs::FundingCreated) -> Result<(), HandleError> { + let _ = self.total_consistency_lock.read().unwrap(); handle_error!(self, self.internal_funding_created(their_node_id, msg), their_node_id) } fn handle_funding_signed(&self, their_node_id: &PublicKey, msg: &msgs::FundingSigned) -> Result<(), HandleError> { + let _ = self.total_consistency_lock.read().unwrap(); handle_error!(self, self.internal_funding_signed(their_node_id, msg), their_node_id) } - 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<(), HandleError> { + let _ = self.total_consistency_lock.read().unwrap(); handle_error!(self, self.internal_funding_locked(their_node_id, msg), their_node_id) } - fn handle_shutdown(&self, their_node_id: &PublicKey, msg: &msgs::Shutdown) -> Result<(Option, Option), HandleError> { + fn handle_shutdown(&self, their_node_id: &PublicKey, msg: &msgs::Shutdown) -> Result<(), HandleError> { + let _ = self.total_consistency_lock.read().unwrap(); handle_error!(self, self.internal_shutdown(their_node_id, msg), their_node_id) } - 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<(), HandleError> { + let _ = self.total_consistency_lock.read().unwrap(); handle_error!(self, self.internal_closing_signed(their_node_id, msg), their_node_id) } fn handle_update_add_htlc(&self, their_node_id: &PublicKey, msg: &msgs::UpdateAddHTLC) -> Result<(), msgs::HandleError> { + let _ = self.total_consistency_lock.read().unwrap(); handle_error!(self, self.internal_update_add_htlc(their_node_id, msg), their_node_id) } fn handle_update_fulfill_htlc(&self, their_node_id: &PublicKey, msg: &msgs::UpdateFulfillHTLC) -> Result<(), HandleError> { + let _ = self.total_consistency_lock.read().unwrap(); handle_error!(self, self.internal_update_fulfill_htlc(their_node_id, msg), their_node_id) } fn handle_update_fail_htlc(&self, their_node_id: &PublicKey, msg: &msgs::UpdateFailHTLC) -> Result<(), HandleError> { + let _ = self.total_consistency_lock.read().unwrap(); handle_error!(self, self.internal_update_fail_htlc(their_node_id, msg), their_node_id) } fn handle_update_fail_malformed_htlc(&self, their_node_id: &PublicKey, msg: &msgs::UpdateFailMalformedHTLC) -> Result<(), HandleError> { + let _ = self.total_consistency_lock.read().unwrap(); handle_error!(self, self.internal_update_fail_malformed_htlc(their_node_id, msg), their_node_id) } - fn handle_commitment_signed(&self, their_node_id: &PublicKey, msg: &msgs::CommitmentSigned) -> Result<(msgs::RevokeAndACK, Option), HandleError> { + fn handle_commitment_signed(&self, their_node_id: &PublicKey, msg: &msgs::CommitmentSigned) -> Result<(), HandleError> { + let _ = self.total_consistency_lock.read().unwrap(); handle_error!(self, self.internal_commitment_signed(their_node_id, msg), their_node_id) } - 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<(), HandleError> { + let _ = self.total_consistency_lock.read().unwrap(); handle_error!(self, self.internal_revoke_and_ack(their_node_id, msg), their_node_id) } fn handle_update_fee(&self, their_node_id: &PublicKey, msg: &msgs::UpdateFee) -> Result<(), HandleError> { + let _ = self.total_consistency_lock.read().unwrap(); handle_error!(self, self.internal_update_fee(their_node_id, msg), their_node_id) } fn handle_announcement_signatures(&self, their_node_id: &PublicKey, msg: &msgs::AnnouncementSignatures) -> Result<(), HandleError> { + let _ = self.total_consistency_lock.read().unwrap(); 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, Option, Option, RAACommitmentOrder), HandleError> { + fn handle_channel_reestablish(&self, their_node_id: &PublicKey, msg: &msgs::ChannelReestablish) -> Result<(), HandleError> { + let _ = self.total_consistency_lock.read().unwrap(); 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 _ = self.total_consistency_lock.read().unwrap(); let mut failed_channels = Vec::new(); let mut failed_payments = Vec::new(); { @@ -2572,9 +2779,11 @@ impl ChannelMessageHandler for ChannelManager { } } - fn peer_connected(&self, their_node_id: &PublicKey) -> Vec { - let mut res = Vec::new(); - let mut channel_state = self.channel_state.lock().unwrap(); + fn peer_connected(&self, their_node_id: &PublicKey) { + 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(); + let pending_msg_events = 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() { @@ -2584,16 +2793,20 @@ impl ChannelMessageHandler for ChannelManager { // drop it. false } else { - res.push(chan.get_channel_reestablish()); + pending_msg_events.push(events::MessageSendEvent::SendChannelReestablish { + node_id: chan.get_their_node_id(), + msg: 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) { + let _ = self.total_consistency_lock.read().unwrap(); + if msg.channel_id == [0; 32] { for chan in self.list_channels() { if chan.remote_network_id == *their_node_id { @@ -2606,15 +2819,397 @@ impl ChannelMessageHandler for ChannelManager { } } +const SERIALIZATION_VERSION: u8 = 1; +const MIN_SERIALIZATION_VERSION: u8 = 1; + +impl Writeable for PendingForwardHTLCInfo { + fn write(&self, writer: &mut W) -> Result<(), ::std::io::Error> { + if let &Some(ref onion) = &self.onion_packet { + 1u8.write(writer)?; + onion.write(writer)?; + } else { + 0u8.write(writer)?; + } + self.incoming_shared_secret.write(writer)?; + self.payment_hash.write(writer)?; + self.short_channel_id.write(writer)?; + self.amt_to_forward.write(writer)?; + self.outgoing_cltv_value.write(writer)?; + Ok(()) + } +} + +impl Readable for PendingForwardHTLCInfo { + fn read(reader: &mut R) -> Result { + let onion_packet = match >::read(reader)? { + 0 => None, + 1 => Some(msgs::OnionPacket::read(reader)?), + _ => return Err(DecodeError::InvalidValue), + }; + Ok(PendingForwardHTLCInfo { + onion_packet, + incoming_shared_secret: Readable::read(reader)?, + payment_hash: Readable::read(reader)?, + short_channel_id: Readable::read(reader)?, + amt_to_forward: Readable::read(reader)?, + outgoing_cltv_value: Readable::read(reader)?, + }) + } +} + +impl Writeable for HTLCFailureMsg { + fn write(&self, writer: &mut W) -> Result<(), ::std::io::Error> { + match self { + &HTLCFailureMsg::Relay(ref fail_msg) => { + 0u8.write(writer)?; + fail_msg.write(writer)?; + }, + &HTLCFailureMsg::Malformed(ref fail_msg) => { + 1u8.write(writer)?; + fail_msg.write(writer)?; + } + } + Ok(()) + } +} + +impl Readable for HTLCFailureMsg { + fn read(reader: &mut R) -> Result { + match >::read(reader)? { + 0 => Ok(HTLCFailureMsg::Relay(Readable::read(reader)?)), + 1 => Ok(HTLCFailureMsg::Malformed(Readable::read(reader)?)), + _ => Err(DecodeError::InvalidValue), + } + } +} + +impl Writeable for PendingHTLCStatus { + fn write(&self, writer: &mut W) -> Result<(), ::std::io::Error> { + match self { + &PendingHTLCStatus::Forward(ref forward_info) => { + 0u8.write(writer)?; + forward_info.write(writer)?; + }, + &PendingHTLCStatus::Fail(ref fail_msg) => { + 1u8.write(writer)?; + fail_msg.write(writer)?; + } + } + Ok(()) + } +} + +impl Readable for PendingHTLCStatus { + fn read(reader: &mut R) -> Result { + match >::read(reader)? { + 0 => Ok(PendingHTLCStatus::Forward(Readable::read(reader)?)), + 1 => Ok(PendingHTLCStatus::Fail(Readable::read(reader)?)), + _ => Err(DecodeError::InvalidValue), + } + } +} + +impl_writeable!(HTLCPreviousHopData, 0, { + short_channel_id, + htlc_id, + incoming_packet_shared_secret +}); + +impl Writeable for HTLCSource { + fn write(&self, writer: &mut W) -> Result<(), ::std::io::Error> { + match self { + &HTLCSource::PreviousHopData(ref hop_data) => { + 0u8.write(writer)?; + hop_data.write(writer)?; + }, + &HTLCSource::OutboundRoute { ref route, ref session_priv, ref first_hop_htlc_msat } => { + 1u8.write(writer)?; + route.write(writer)?; + session_priv.write(writer)?; + first_hop_htlc_msat.write(writer)?; + } + } + Ok(()) + } +} + +impl Readable for HTLCSource { + fn read(reader: &mut R) -> Result { + match >::read(reader)? { + 0 => Ok(HTLCSource::PreviousHopData(Readable::read(reader)?)), + 1 => Ok(HTLCSource::OutboundRoute { + route: Readable::read(reader)?, + session_priv: Readable::read(reader)?, + first_hop_htlc_msat: Readable::read(reader)?, + }), + _ => Err(DecodeError::InvalidValue), + } + } +} + +impl Writeable for HTLCFailReason { + fn write(&self, writer: &mut W) -> Result<(), ::std::io::Error> { + match self { + &HTLCFailReason::ErrorPacket { ref err } => { + 0u8.write(writer)?; + err.write(writer)?; + }, + &HTLCFailReason::Reason { ref failure_code, ref data } => { + 1u8.write(writer)?; + failure_code.write(writer)?; + data.write(writer)?; + } + } + Ok(()) + } +} + +impl Readable for HTLCFailReason { + fn read(reader: &mut R) -> Result { + match >::read(reader)? { + 0 => Ok(HTLCFailReason::ErrorPacket { err: Readable::read(reader)? }), + 1 => Ok(HTLCFailReason::Reason { + failure_code: Readable::read(reader)?, + data: Readable::read(reader)?, + }), + _ => Err(DecodeError::InvalidValue), + } + } +} + +impl_writeable!(HTLCForwardInfo, 0, { + prev_short_channel_id, + prev_htlc_id, + forward_info +}); + +impl Writeable for ChannelManager { + fn write(&self, writer: &mut W) -> Result<(), ::std::io::Error> { + let _ = self.total_consistency_lock.write().unwrap(); + + writer.write_all(&[SERIALIZATION_VERSION; 1])?; + writer.write_all(&[MIN_SERIALIZATION_VERSION; 1])?; + + self.genesis_hash.write(writer)?; + (self.latest_block_height.load(Ordering::Acquire) as u32).write(writer)?; + self.last_block_hash.lock().unwrap().write(writer)?; + + let channel_state = self.channel_state.lock().unwrap(); + let mut unfunded_channels = 0; + for (_, channel) in channel_state.by_id.iter() { + if !channel.is_funding_initiated() { + unfunded_channels += 1; + } + } + ((channel_state.by_id.len() - unfunded_channels) as u64).write(writer)?; + for (_, channel) in channel_state.by_id.iter() { + if channel.is_funding_initiated() { + channel.write(writer)?; + } + } + + (channel_state.forward_htlcs.len() as u64).write(writer)?; + for (short_channel_id, pending_forwards) in channel_state.forward_htlcs.iter() { + short_channel_id.write(writer)?; + (pending_forwards.len() as u64).write(writer)?; + for forward in pending_forwards { + forward.write(writer)?; + } + } + + (channel_state.claimable_htlcs.len() as u64).write(writer)?; + for (payment_hash, previous_hops) in channel_state.claimable_htlcs.iter() { + payment_hash.write(writer)?; + (previous_hops.len() as u64).write(writer)?; + for previous_hop in previous_hops { + previous_hop.write(writer)?; + } + } + + Ok(()) + } +} + +/// Arguments for the creation of a ChannelManager that are not deserialized. +/// +/// At a high-level, the process for deserializing a ChannelManager and resuming normal operation +/// is: +/// 1) Deserialize all stored ChannelMonitors. +/// 2) Deserialize the ChannelManager by filling in this struct and calling <(Sha256dHash, +/// ChannelManager)>::read(reader, args). +/// This may result in closing some Channels if the ChannelMonitor is newer than the stored +/// ChannelManager state to ensure no loss of funds. Thus, transactions may be broadcasted. +/// 3) Register all relevant ChannelMonitor outpoints with your chain watch mechanism using +/// ChannelMonitor::get_monitored_outpoints and ChannelMonitor::get_funding_txo(). +/// 4) Reconnect blocks on your ChannelMonitors. +/// 5) Move the ChannelMonitors into your local ManyChannelMonitor. +/// 6) Disconnect/connect blocks on the ChannelManager. +/// 7) Register the new ChannelManager with your ChainWatchInterface (this does not happen +/// automatically as it does in ChannelManager::new()). +pub struct ChannelManagerReadArgs<'a> { + /// The keys provider which will give us relevant keys. Some keys will be loaded during + /// deserialization. + pub keys_manager: Arc, + + /// The fee_estimator for use in the ChannelManager in the future. + /// + /// No calls to the FeeEstimator will be made during deserialization. + pub fee_estimator: Arc, + /// The ManyChannelMonitor for use in the ChannelManager in the future. + /// + /// 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, + /// The ChainWatchInterface for use in the ChannelManager in the future. + /// + /// No calls to the ChainWatchInterface will be made during deserialization. + pub chain_monitor: Arc, + /// 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 + /// force-closed during deserialization. + pub tx_broadcaster: Arc, + /// The Logger for use in the ChannelManager and which may be used to log information during + /// deserialization. + pub logger: Arc, + /// Default settings used for new channels. Any existing channels will continue to use the + /// runtime settings which were stored when the ChannelManager was serialized. + pub default_config: UserConfig, + + /// A map from channel funding outpoints to ChannelMonitors for those channels (ie + /// value.get_funding_txo() should be the key). + /// + /// If a monitor is inconsistent with the channel state during deserialization the channel will + /// be force-closed using the data in the channelmonitor and the Channel will be dropped. This + /// is true for missing channels as well. If there is a monitor missing for which we find + /// channel data Err(DecodeError::InvalidValue) will be returned. + /// + /// In such cases the latest local transactions will be sent to the tx_broadcaster included in + /// this struct. + pub channel_monitors: &'a HashMap, +} + +impl<'a, R : ::std::io::Read> ReadableArgs> for (Sha256dHash, ChannelManager) { + fn read(reader: &mut R, args: ChannelManagerReadArgs<'a>) -> Result { + let _ver: u8 = Readable::read(reader)?; + let min_ver: u8 = Readable::read(reader)?; + if min_ver > SERIALIZATION_VERSION { + return Err(DecodeError::UnknownVersion); + } + + let genesis_hash: Sha256dHash = Readable::read(reader)?; + let latest_block_height: u32 = Readable::read(reader)?; + let last_block_hash: Sha256dHash = Readable::read(reader)?; + + let mut closed_channels = Vec::new(); + + let channel_count: u64 = Readable::read(reader)?; + let mut funding_txo_set = HashSet::with_capacity(cmp::min(channel_count as usize, 128)); + 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())?; + 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 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() { + let mut force_close_res = channel.force_shutdown(); + force_close_res.0 = monitor.get_latest_local_commitment_txn(); + closed_channels.push(force_close_res); + } else { + if let Some(short_channel_id) = channel.get_short_channel_id() { + short_to_id.insert(short_channel_id, channel.channel_id()); + } + by_id.insert(channel.channel_id(), channel); + } + } else { + return Err(DecodeError::InvalidValue); + } + } + + for (ref funding_txo, ref monitor) in args.channel_monitors.iter() { + if !funding_txo_set.contains(funding_txo) { + closed_channels.push((monitor.get_latest_local_commitment_txn(), Vec::new())); + } + } + + let forward_htlcs_count: u64 = Readable::read(reader)?; + let mut forward_htlcs = HashMap::with_capacity(cmp::min(forward_htlcs_count as usize, 128)); + for _ in 0..forward_htlcs_count { + let short_channel_id = Readable::read(reader)?; + let pending_forwards_count: u64 = Readable::read(reader)?; + let mut pending_forwards = Vec::with_capacity(cmp::min(pending_forwards_count as usize, 128)); + for _ in 0..pending_forwards_count { + pending_forwards.push(Readable::read(reader)?); + } + forward_htlcs.insert(short_channel_id, pending_forwards); + } + + let claimable_htlcs_count: u64 = Readable::read(reader)?; + let mut claimable_htlcs = HashMap::with_capacity(cmp::min(claimable_htlcs_count as usize, 128)); + for _ in 0..claimable_htlcs_count { + let payment_hash = Readable::read(reader)?; + let previous_hops_len: u64 = Readable::read(reader)?; + let mut previous_hops = Vec::with_capacity(cmp::min(previous_hops_len as usize, 2)); + for _ in 0..previous_hops_len { + previous_hops.push(Readable::read(reader)?); + } + claimable_htlcs.insert(payment_hash, previous_hops); + } + + let channel_manager = ChannelManager { + genesis_hash, + fee_estimator: args.fee_estimator, + monitor: args.monitor, + chain_monitor: args.chain_monitor, + tx_broadcaster: args.tx_broadcaster, + + latest_block_height: AtomicUsize::new(latest_block_height as usize), + last_block_hash: Mutex::new(last_block_hash), + secp_ctx: Secp256k1::new(), + + channel_state: Mutex::new(ChannelHolder { + by_id, + short_to_id, + next_forward: Instant::now(), + forward_htlcs, + claimable_htlcs, + pending_msg_events: Vec::new(), + }), + our_network_key: args.keys_manager.get_node_secret(), + + pending_events: Mutex::new(Vec::new()), + total_consistency_lock: RwLock::new(()), + keys_manager: args.keys_manager, + logger: args.logger, + default_configuration: args.default_config, + }; + + for close_res in closed_channels.drain(..) { + channel_manager.finish_force_close_channel(close_res); + //TODO: Broadcast channel update for closed channels, but only after we've made a + //connection or two. + } + + Ok((last_block_hash.clone(), channel_manager)) + } +} + #[cfg(test)] mod tests { use chain::chaininterface; use chain::transaction::OutPoint; - use chain::chaininterface::ChainListener; + use chain::chaininterface::{ChainListener, ChainWatchInterface}; use chain::keysinterface::KeysInterface; use chain::keysinterface; - use ln::channelmanager::{ChannelManager,OnionKeys,PaymentFailReason}; - use ln::channelmonitor::{ChannelMonitorUpdateErr, CLTV_CLAIM_BUFFER, HTLC_FAIL_TIMEOUT_BLOCKS}; + use ln::channelmanager::{ChannelManager,ChannelManagerReadArgs,OnionKeys,PaymentFailReason,RAACommitmentOrder}; + use ln::channelmonitor::{ChannelMonitor, ChannelMonitorUpdateErr, CLTV_CLAIM_BUFFER, HTLC_FAIL_TIMEOUT_BLOCKS, ManyChannelMonitor}; use ln::router::{Route, RouteHop, Router}; use ln::msgs; use ln::msgs::{ChannelMessageHandler,RoutingMessageHandler}; @@ -2622,7 +3217,8 @@ mod tests { use util::events::{Event, EventsProvider, MessageSendEvent, MessageSendEventsProvider}; use util::errors::APIError; use util::logger::Logger; - use util::ser::Writeable; + use util::ser::{Writeable, Writer, ReadableArgs}; + use util::config::UserConfig; use bitcoin::util::hash::Sha256dHash; use bitcoin::blockdata::block::{Block, BlockHeader}; @@ -2782,22 +3378,22 @@ mod tests { // Returning Errors test vectors from BOLT 4 let onion_keys = build_test_onion_keys(); - let onion_error = ChannelManager::build_failure_packet(&onion_keys[4].shared_secret, 0x2002, &[0; 0]); + let onion_error = ChannelManager::build_failure_packet(&onion_keys[4].shared_secret[..], 0x2002, &[0; 0]); assert_eq!(onion_error.encode(), hex::decode("4c2fc8bc08510334b6833ad9c3e79cd1b52ae59dfe5c2a4b23ead50f09f7ee0b0002200200fe0000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000").unwrap()); - let onion_packet_1 = ChannelManager::encrypt_failure_packet(&onion_keys[4].shared_secret, &onion_error.encode()[..]); + let onion_packet_1 = ChannelManager::encrypt_failure_packet(&onion_keys[4].shared_secret[..], &onion_error.encode()[..]); assert_eq!(onion_packet_1.data, hex::decode("a5e6bd0c74cb347f10cce367f949098f2457d14c046fd8a22cb96efb30b0fdcda8cb9168b50f2fd45edd73c1b0c8b33002df376801ff58aaa94000bf8a86f92620f343baef38a580102395ae3abf9128d1047a0736ff9b83d456740ebbb4aeb3aa9737f18fb4afb4aa074fb26c4d702f42968888550a3bded8c05247e045b866baef0499f079fdaeef6538f31d44deafffdfd3afa2fb4ca9082b8f1c465371a9894dd8c243fb4847e004f5256b3e90e2edde4c9fb3082ddfe4d1e734cacd96ef0706bf63c9984e22dc98851bcccd1c3494351feb458c9c6af41c0044bea3c47552b1d992ae542b17a2d0bba1a096c78d169034ecb55b6e3a7263c26017f033031228833c1daefc0dedb8cf7c3e37c9c37ebfe42f3225c326e8bcfd338804c145b16e34e4").unwrap()); - let onion_packet_2 = ChannelManager::encrypt_failure_packet(&onion_keys[3].shared_secret, &onion_packet_1.data[..]); + let onion_packet_2 = ChannelManager::encrypt_failure_packet(&onion_keys[3].shared_secret[..], &onion_packet_1.data[..]); assert_eq!(onion_packet_2.data, hex::decode("c49a1ce81680f78f5f2000cda36268de34a3f0a0662f55b4e837c83a8773c22aa081bab1616a0011585323930fa5b9fae0c85770a2279ff59ec427ad1bbff9001c0cd1497004bd2a0f68b50704cf6d6a4bf3c8b6a0833399a24b3456961ba00736785112594f65b6b2d44d9f5ea4e49b5e1ec2af978cbe31c67114440ac51a62081df0ed46d4a3df295da0b0fe25c0115019f03f15ec86fabb4c852f83449e812f141a9395b3f70b766ebbd4ec2fae2b6955bd8f32684c15abfe8fd3a6261e52650e8807a92158d9f1463261a925e4bfba44bd20b166d532f0017185c3a6ac7957adefe45559e3072c8dc35abeba835a8cb01a71a15c736911126f27d46a36168ca5ef7dccd4e2886212602b181463e0dd30185c96348f9743a02aca8ec27c0b90dca270").unwrap()); - let onion_packet_3 = ChannelManager::encrypt_failure_packet(&onion_keys[2].shared_secret, &onion_packet_2.data[..]); + let onion_packet_3 = ChannelManager::encrypt_failure_packet(&onion_keys[2].shared_secret[..], &onion_packet_2.data[..]); assert_eq!(onion_packet_3.data, hex::decode("a5d3e8634cfe78b2307d87c6d90be6fe7855b4f2cc9b1dfb19e92e4b79103f61ff9ac25f412ddfb7466e74f81b3e545563cdd8f5524dae873de61d7bdfccd496af2584930d2b566b4f8d3881f8c043df92224f38cf094cfc09d92655989531524593ec6d6caec1863bdfaa79229b5020acc034cd6deeea1021c50586947b9b8e6faa83b81fbfa6133c0af5d6b07c017f7158fa94f0d206baf12dda6b68f785b773b360fd0497e16cc402d779c8d48d0fa6315536ef0660f3f4e1865f5b38ea49c7da4fd959de4e83ff3ab686f059a45c65ba2af4a6a79166aa0f496bf04d06987b6d2ea205bdb0d347718b9aeff5b61dfff344993a275b79717cd815b6ad4c0beb568c4ac9c36ff1c315ec1119a1993c4b61e6eaa0375e0aaf738ac691abd3263bf937e3").unwrap()); - let onion_packet_4 = ChannelManager::encrypt_failure_packet(&onion_keys[1].shared_secret, &onion_packet_3.data[..]); + let onion_packet_4 = ChannelManager::encrypt_failure_packet(&onion_keys[1].shared_secret[..], &onion_packet_3.data[..]); assert_eq!(onion_packet_4.data, hex::decode("aac3200c4968f56b21f53e5e374e3a2383ad2b1b6501bbcc45abc31e59b26881b7dfadbb56ec8dae8857add94e6702fb4c3a4de22e2e669e1ed926b04447fc73034bb730f4932acd62727b75348a648a1128744657ca6a4e713b9b646c3ca66cac02cdab44dd3439890ef3aaf61708714f7375349b8da541b2548d452d84de7084bb95b3ac2345201d624d31f4d52078aa0fa05a88b4e20202bd2b86ac5b52919ea305a8949de95e935eed0319cf3cf19ebea61d76ba92532497fcdc9411d06bcd4275094d0a4a3c5d3a945e43305a5a9256e333e1f64dbca5fcd4e03a39b9012d197506e06f29339dfee3331995b21615337ae060233d39befea925cc262873e0530408e6990f1cbd233a150ef7b004ff6166c70c68d9f8c853c1abca640b8660db2921").unwrap()); - let onion_packet_5 = ChannelManager::encrypt_failure_packet(&onion_keys[0].shared_secret, &onion_packet_4.data[..]); + let onion_packet_5 = ChannelManager::encrypt_failure_packet(&onion_keys[0].shared_secret[..], &onion_packet_4.data[..]); assert_eq!(onion_packet_5.data, hex::decode("9c5add3963fc7f6ed7f148623c84134b5647e1306419dbe2174e523fa9e2fbed3a06a19f899145610741c83ad40b7712aefaddec8c6baf7325d92ea4ca4d1df8bce517f7e54554608bf2bd8071a4f52a7a2f7ffbb1413edad81eeea5785aa9d990f2865dc23b4bc3c301a94eec4eabebca66be5cf638f693ec256aec514620cc28ee4a94bd9565bc4d4962b9d3641d4278fb319ed2b84de5b665f307a2db0f7fbb757366067d88c50f7e829138fde4f78d39b5b5802f1b92a8a820865af5cc79f9f30bc3f461c66af95d13e5e1f0381c184572a91dee1c849048a647a1158cf884064deddbf1b0b88dfe2f791428d0ba0f6fb2f04e14081f69165ae66d9297c118f0907705c9c4954a199bae0bb96fad763d690e7daa6cfda59ba7f2c8d11448b604d12d").unwrap()); } @@ -2817,6 +3413,7 @@ mod tests { chan_monitor: Arc, node: Arc, router: Router, + node_seed: [u8; 32], network_payment_count: Rc>, network_chan_count: Rc>, } @@ -2831,30 +3428,79 @@ mod tests { } } - fn create_chan_between_nodes(node_a: &Node, node_b: &Node) -> (msgs::ChannelAnnouncement, msgs::ChannelUpdate, msgs::ChannelUpdate, [u8; 32], Transaction) { - create_chan_between_nodes_with_value(node_a, node_b, 100000, 10001) - } - - fn create_chan_between_nodes_with_value(node_a: &Node, node_b: &Node, channel_value: u64, push_msat: u64) -> (msgs::ChannelAnnouncement, msgs::ChannelUpdate, msgs::ChannelUpdate, [u8; 32], Transaction) { - let (funding_locked, channel_id, tx) = create_chan_between_nodes_with_value_a(node_a, node_b, channel_value, push_msat); - let (announcement, as_update, bs_update) = create_chan_between_nodes_with_value_b(node_a, node_b, &funding_locked); - (announcement, as_update, bs_update, channel_id, tx) + fn create_chan_between_nodes(node_a: &Node, node_b: &Node) -> (msgs::ChannelAnnouncement, msgs::ChannelUpdate, msgs::ChannelUpdate, [u8; 32], Transaction) { + create_chan_between_nodes_with_value(node_a, node_b, 100000, 10001) + } + + fn create_chan_between_nodes_with_value(node_a: &Node, node_b: &Node, channel_value: u64, push_msat: u64) -> (msgs::ChannelAnnouncement, msgs::ChannelUpdate, msgs::ChannelUpdate, [u8; 32], Transaction) { + let (funding_locked, channel_id, tx) = create_chan_between_nodes_with_value_a(node_a, node_b, channel_value, push_msat); + let (announcement, as_update, bs_update) = create_chan_between_nodes_with_value_b(node_a, node_b, &funding_locked); + (announcement, as_update, bs_update, channel_id, tx) + } + + macro_rules! get_revoke_commit_msgs { + ($node: expr, $node_id: expr) => { + { + let events = $node.node.get_and_clear_pending_msg_events(); + assert_eq!(events.len(), 2); + (match events[0] { + MessageSendEvent::SendRevokeAndACK { ref node_id, ref msg } => { + assert_eq!(*node_id, $node_id); + (*msg).clone() + }, + _ => panic!("Unexpected event"), + }, match events[1] { + MessageSendEvent::UpdateHTLCs { ref node_id, ref updates } => { + assert_eq!(*node_id, $node_id); + assert!(updates.update_add_htlcs.is_empty()); + assert!(updates.update_fulfill_htlcs.is_empty()); + assert!(updates.update_fail_htlcs.is_empty()); + assert!(updates.update_fail_malformed_htlcs.is_empty()); + assert!(updates.update_fee.is_none()); + updates.commitment_signed.clone() + }, + _ => panic!("Unexpected event"), + }) + } + } + } + + macro_rules! get_event_msg { + ($node: expr, $event_type: path, $node_id: expr) => { + { + let events = $node.node.get_and_clear_pending_msg_events(); + assert_eq!(events.len(), 1); + match events[0] { + $event_type { ref node_id, ref msg } => { + assert_eq!(*node_id, $node_id); + (*msg).clone() + }, + _ => panic!("Unexpected event"), + } + } + } + } + + macro_rules! get_htlc_update_msgs { + ($node: expr, $node_id: expr) => { + { + let events = $node.node.get_and_clear_pending_msg_events(); + assert_eq!(events.len(), 1); + match events[0] { + MessageSendEvent::UpdateHTLCs { ref node_id, ref updates } => { + assert_eq!(*node_id, $node_id); + (*updates).clone() + }, + _ => panic!("Unexpected event"), + } + } + } } fn create_chan_between_nodes_with_value_init(node_a: &Node, node_b: &Node, channel_value: u64, push_msat: u64) -> Transaction { node_a.node.create_channel(node_b.node.get_our_node_id(), channel_value, push_msat, 42).unwrap(); - - let events_1 = node_a.node.get_and_clear_pending_msg_events(); - assert_eq!(events_1.len(), 1); - let accept_chan = match events_1[0] { - MessageSendEvent::SendOpenChannel { ref node_id, ref msg } => { - assert_eq!(*node_id, node_b.node.get_our_node_id()); - node_b.node.handle_open_channel(&node_a.node.get_our_node_id(), msg).unwrap() - }, - _ => panic!("Unexpected event"), - }; - - node_a.node.handle_accept_channel(&node_b.node.get_our_node_id(), &accept_chan).unwrap(); + node_b.node.handle_open_channel(&node_a.node.get_our_node_id(), &get_event_msg!(node_a, MessageSendEvent::SendOpenChannel, node_b.node.get_our_node_id())).unwrap(); + node_a.node.handle_accept_channel(&node_b.node.get_our_node_id(), &get_event_msg!(node_b, MessageSendEvent::SendAcceptChannel, node_a.node.get_our_node_id())).unwrap(); let chan_id = *node_a.network_chan_count.borrow(); let tx; @@ -2881,22 +3527,15 @@ mod tests { _ => panic!("Unexpected event"), } - let events_3 = node_a.node.get_and_clear_pending_msg_events(); - assert_eq!(events_3.len(), 1); - let funding_signed = match events_3[0] { - MessageSendEvent::SendFundingCreated { ref node_id, ref msg } => { - assert_eq!(*node_id, node_b.node.get_our_node_id()); - let res = node_b.node.handle_funding_created(&node_a.node.get_our_node_id(), msg).unwrap(); - let mut added_monitors = node_b.chan_monitor.added_monitors.lock().unwrap(); - assert_eq!(added_monitors.len(), 1); - assert_eq!(added_monitors[0].0, funding_output); - added_monitors.clear(); - res - }, - _ => panic!("Unexpected event"), - }; + node_b.node.handle_funding_created(&node_a.node.get_our_node_id(), &get_event_msg!(node_a, MessageSendEvent::SendFundingCreated, node_b.node.get_our_node_id())).unwrap(); + { + let mut added_monitors = node_b.chan_monitor.added_monitors.lock().unwrap(); + assert_eq!(added_monitors.len(), 1); + assert_eq!(added_monitors[0].0, funding_output); + added_monitors.clear(); + } - node_a.node.handle_funding_signed(&node_b.node.get_our_node_id(), &funding_signed).unwrap(); + node_a.node.handle_funding_signed(&node_b.node.get_our_node_id(), &get_event_msg!(node_b, MessageSendEvent::SendFundingSigned, node_a.node.get_our_node_id())).unwrap(); { let mut added_monitors = node_a.chan_monitor.added_monitors.lock().unwrap(); assert_eq!(added_monitors.len(), 1); @@ -2919,30 +3558,27 @@ mod tests { fn create_chan_between_nodes_with_value_confirm(node_a: &Node, node_b: &Node, tx: &Transaction) -> ((msgs::FundingLocked, msgs::AnnouncementSignatures), [u8; 32]) { confirm_transaction(&node_b.chain_monitor, &tx, tx.version); - let events_5 = node_b.node.get_and_clear_pending_msg_events(); - assert_eq!(events_5.len(), 1); - match events_5[0] { - MessageSendEvent::SendFundingLocked { ref node_id, ref msg, ref announcement_sigs } => { - assert_eq!(*node_id, node_a.node.get_our_node_id()); - assert!(announcement_sigs.is_none()); - node_a.node.handle_funding_locked(&node_b.node.get_our_node_id(), msg).unwrap() - }, - _ => panic!("Unexpected event"), - }; + node_a.node.handle_funding_locked(&node_b.node.get_our_node_id(), &get_event_msg!(node_b, MessageSendEvent::SendFundingLocked, node_a.node.get_our_node_id())).unwrap(); let channel_id; confirm_transaction(&node_a.chain_monitor, &tx, tx.version); let events_6 = node_a.node.get_and_clear_pending_msg_events(); - assert_eq!(events_6.len(), 1); - (match events_6[0] { - MessageSendEvent::SendFundingLocked { ref node_id, ref msg, ref announcement_sigs } => { + assert_eq!(events_6.len(), 2); + ((match events_6[0] { + MessageSendEvent::SendFundingLocked { ref node_id, ref msg } => { channel_id = msg.channel_id.clone(); assert_eq!(*node_id, node_b.node.get_our_node_id()); - (msg.clone(), announcement_sigs.clone().unwrap()) + msg.clone() + }, + _ => panic!("Unexpected event"), + }, match events_6[1] { + MessageSendEvent::SendAnnouncementSignatures { ref node_id, ref msg } => { + assert_eq!(*node_id, node_b.node.get_our_node_id()); + msg.clone() }, _ => panic!("Unexpected event"), - }, channel_id) + }), channel_id) } fn create_chan_between_nodes_with_value_a(node_a: &Node, node_b: &Node, channel_value: u64, push_msat: u64) -> ((msgs::FundingLocked, msgs::AnnouncementSignatures), [u8; 32], Transaction) { @@ -2952,11 +3588,9 @@ mod tests { } fn create_chan_between_nodes_with_value_b(node_a: &Node, node_b: &Node, as_funding_msgs: &(msgs::FundingLocked, msgs::AnnouncementSignatures)) -> (msgs::ChannelAnnouncement, msgs::ChannelUpdate, msgs::ChannelUpdate) { - let bs_announcement_sigs = { - let bs_announcement_sigs = node_b.node.handle_funding_locked(&node_a.node.get_our_node_id(), &as_funding_msgs.0).unwrap().unwrap(); - node_b.node.handle_announcement_signatures(&node_a.node.get_our_node_id(), &as_funding_msgs.1).unwrap(); - bs_announcement_sigs - }; + node_b.node.handle_funding_locked(&node_a.node.get_our_node_id(), &as_funding_msgs.0).unwrap(); + let bs_announcement_sigs = get_event_msg!(node_b, MessageSendEvent::SendAnnouncementSignatures, node_a.node.get_our_node_id()); + node_b.node.handle_announcement_signatures(&node_a.node.get_our_node_id(), &as_funding_msgs.1).unwrap(); let events_7 = node_b.node.get_and_clear_pending_msg_events(); assert_eq!(events_7.len(), 1); @@ -3009,68 +3643,91 @@ mod tests { } fn close_channel(outbound_node: &Node, inbound_node: &Node, channel_id: &[u8; 32], funding_tx: Transaction, close_inbound_first: bool) -> (msgs::ChannelUpdate, msgs::ChannelUpdate) { - let (node_a, broadcaster_a) = if close_inbound_first { (&inbound_node.node, &inbound_node.tx_broadcaster) } else { (&outbound_node.node, &outbound_node.tx_broadcaster) }; + let (node_a, broadcaster_a, struct_a) = if close_inbound_first { (&inbound_node.node, &inbound_node.tx_broadcaster, inbound_node) } else { (&outbound_node.node, &outbound_node.tx_broadcaster, outbound_node) }; let (node_b, broadcaster_b) = if close_inbound_first { (&outbound_node.node, &outbound_node.tx_broadcaster) } else { (&inbound_node.node, &inbound_node.tx_broadcaster) }; let (tx_a, tx_b); node_a.close_channel(channel_id).unwrap(); - let events_1 = node_a.get_and_clear_pending_msg_events(); - assert_eq!(events_1.len(), 1); - let shutdown_a = match events_1[0] { + node_b.handle_shutdown(&node_a.get_our_node_id(), &get_event_msg!(struct_a, MessageSendEvent::SendShutdown, node_b.get_our_node_id())).unwrap(); + + let events_1 = node_b.get_and_clear_pending_msg_events(); + assert!(events_1.len() >= 1); + let shutdown_b = match events_1[0] { MessageSendEvent::SendShutdown { ref node_id, ref msg } => { - assert_eq!(node_id, &node_b.get_our_node_id()); + assert_eq!(node_id, &node_a.get_our_node_id()); msg.clone() }, _ => panic!("Unexpected event"), }; - let (shutdown_b, mut closing_signed_b) = node_b.handle_shutdown(&node_a.get_our_node_id(), &shutdown_a).unwrap(); - if !close_inbound_first { - assert!(closing_signed_b.is_none()); + let closing_signed_b = if !close_inbound_first { + assert_eq!(events_1.len(), 1); + None + } else { + Some(match events_1[1] { + MessageSendEvent::SendClosingSigned { ref node_id, ref msg } => { + assert_eq!(node_id, &node_a.get_our_node_id()); + msg.clone() + }, + _ => panic!("Unexpected event"), + }) + }; + + macro_rules! get_closing_signed_broadcast { + ($node: expr, $dest_pubkey: expr) => { + { + let events = $node.get_and_clear_pending_msg_events(); + assert!(events.len() == 1 || events.len() == 2); + (match events[events.len() - 1] { + MessageSendEvent::BroadcastChannelUpdate { ref msg } => { + msg.clone() + }, + _ => panic!("Unexpected event"), + }, if events.len() == 2 { + match events[0] { + MessageSendEvent::SendClosingSigned { ref node_id, ref msg } => { + assert_eq!(*node_id, $dest_pubkey); + Some(msg.clone()) + }, + _ => panic!("Unexpected event"), + } + } else { None }) + } + } } - let (empty_a, mut closing_signed_a) = node_a.handle_shutdown(&node_b.get_our_node_id(), &shutdown_b.unwrap()).unwrap(); - assert!(empty_a.is_none()); - if close_inbound_first { - assert!(closing_signed_a.is_none()); - closing_signed_a = node_a.handle_closing_signed(&node_b.get_our_node_id(), &closing_signed_b.unwrap()).unwrap(); + + node_a.handle_shutdown(&node_b.get_our_node_id(), &shutdown_b).unwrap(); + let (as_update, bs_update) = if close_inbound_first { + assert!(node_a.get_and_clear_pending_msg_events().is_empty()); + node_a.handle_closing_signed(&node_b.get_our_node_id(), &closing_signed_b.unwrap()).unwrap(); assert_eq!(broadcaster_a.txn_broadcasted.lock().unwrap().len(), 1); tx_a = broadcaster_a.txn_broadcasted.lock().unwrap().remove(0); + let (as_update, closing_signed_a) = get_closing_signed_broadcast!(node_a, node_b.get_our_node_id()); - let empty_b = node_b.handle_closing_signed(&node_a.get_our_node_id(), &closing_signed_a.unwrap()).unwrap(); - assert!(empty_b.is_none()); + node_b.handle_closing_signed(&node_a.get_our_node_id(), &closing_signed_a.unwrap()).unwrap(); + let (bs_update, none_b) = get_closing_signed_broadcast!(node_b, node_a.get_our_node_id()); + assert!(none_b.is_none()); assert_eq!(broadcaster_b.txn_broadcasted.lock().unwrap().len(), 1); tx_b = broadcaster_b.txn_broadcasted.lock().unwrap().remove(0); + (as_update, bs_update) } else { - closing_signed_b = node_b.handle_closing_signed(&node_a.get_our_node_id(), &closing_signed_a.unwrap()).unwrap(); + let closing_signed_a = get_event_msg!(struct_a, MessageSendEvent::SendClosingSigned, node_b.get_our_node_id()); + + node_b.handle_closing_signed(&node_a.get_our_node_id(), &closing_signed_a).unwrap(); assert_eq!(broadcaster_b.txn_broadcasted.lock().unwrap().len(), 1); tx_b = broadcaster_b.txn_broadcasted.lock().unwrap().remove(0); + let (bs_update, closing_signed_b) = get_closing_signed_broadcast!(node_b, node_a.get_our_node_id()); - let empty_a2 = node_a.handle_closing_signed(&node_b.get_our_node_id(), &closing_signed_b.unwrap()).unwrap(); - assert!(empty_a2.is_none()); + node_a.handle_closing_signed(&node_b.get_our_node_id(), &closing_signed_b.unwrap()).unwrap(); + let (as_update, none_a) = get_closing_signed_broadcast!(node_a, node_b.get_our_node_id()); + assert!(none_a.is_none()); assert_eq!(broadcaster_a.txn_broadcasted.lock().unwrap().len(), 1); tx_a = broadcaster_a.txn_broadcasted.lock().unwrap().remove(0); - } + (as_update, bs_update) + }; assert_eq!(tx_a, tx_b); check_spends!(tx_a, funding_tx); - let events_2 = node_a.get_and_clear_pending_msg_events(); - assert_eq!(events_2.len(), 1); - let as_update = match events_2[0] { - MessageSendEvent::BroadcastChannelUpdate { ref msg } => { - msg.clone() - }, - _ => panic!("Unexpected event"), - }; - - let events_3 = node_b.get_and_clear_pending_msg_events(); - assert_eq!(events_3.len(), 1); - let bs_update = match events_3[0] { - MessageSendEvent::BroadcastChannelUpdate { ref msg } => { - msg.clone() - }, - _ => panic!("Unexpected event"), - }; - (as_update, bs_update) } @@ -3110,19 +3767,32 @@ mod tests { ($node_a: expr, $node_b: expr, $commitment_signed: expr, $fail_backwards: expr) => { { check_added_monitors!($node_a, 0); - let (as_revoke_and_ack, as_commitment_signed) = $node_a.node.handle_commitment_signed(&$node_b.node.get_our_node_id(), &$commitment_signed).unwrap(); + assert!($node_a.node.get_and_clear_pending_msg_events().is_empty()); + $node_a.node.handle_commitment_signed(&$node_b.node.get_our_node_id(), &$commitment_signed).unwrap(); + let (as_revoke_and_ack, as_commitment_signed) = get_revoke_commit_msgs!($node_a, $node_b.node.get_our_node_id()); check_added_monitors!($node_a, 1); check_added_monitors!($node_b, 0); - assert!($node_b.node.handle_revoke_and_ack(&$node_a.node.get_our_node_id(), &as_revoke_and_ack).unwrap().is_none()); + assert!($node_b.node.get_and_clear_pending_msg_events().is_empty()); + $node_b.node.handle_revoke_and_ack(&$node_a.node.get_our_node_id(), &as_revoke_and_ack).unwrap(); + assert!($node_b.node.get_and_clear_pending_msg_events().is_empty()); check_added_monitors!($node_b, 1); - let (bs_revoke_and_ack, bs_none) = $node_b.node.handle_commitment_signed(&$node_a.node.get_our_node_id(), &as_commitment_signed.unwrap()).unwrap(); - assert!(bs_none.is_none()); + $node_b.node.handle_commitment_signed(&$node_a.node.get_our_node_id(), &as_commitment_signed).unwrap(); + let bs_revoke_and_ack = get_event_msg!($node_b, MessageSendEvent::SendRevokeAndACK, $node_a.node.get_our_node_id()); check_added_monitors!($node_b, 1); if $fail_backwards { assert!($node_a.node.get_and_clear_pending_events().is_empty()); assert!($node_a.node.get_and_clear_pending_msg_events().is_empty()); } - assert!($node_a.node.handle_revoke_and_ack(&$node_b.node.get_our_node_id(), &bs_revoke_and_ack).unwrap().is_none()); + $node_a.node.handle_revoke_and_ack(&$node_b.node.get_our_node_id(), &bs_revoke_and_ack).unwrap(); + if $fail_backwards { + let channel_state = $node_a.node.channel_state.lock().unwrap(); + assert_eq!(channel_state.pending_msg_events.len(), 1); + if let MessageSendEvent::UpdateHTLCs { ref node_id, .. } = channel_state.pending_msg_events[0] { + assert_ne!(*node_id, $node_b.node.get_our_node_id()); + } else { panic!("Unexpected event"); } + } else { + assert!($node_a.node.get_and_clear_pending_msg_events().is_empty()); + } { let mut added_monitors = $node_a.chan_monitor.added_monitors.lock().unwrap(); if $fail_backwards { @@ -3210,45 +3880,65 @@ mod tests { check_added_monitors!(expected_route.last().unwrap(), 1); let mut next_msgs: Option<(msgs::UpdateFulfillHTLC, msgs::CommitmentSigned)> = None; - macro_rules! update_fulfill_dance { - ($node: expr, $prev_node: expr, $last_node: expr) => { + let mut expected_next_node = expected_route.last().unwrap().node.get_our_node_id(); + macro_rules! get_next_msgs { + ($node: expr) => { + { + let events = $node.node.get_and_clear_pending_msg_events(); + assert_eq!(events.len(), 1); + match events[0] { + MessageSendEvent::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 update_fee, 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()); + assert!(update_fee.is_none()); + expected_next_node = node_id.clone(); + Some((update_fulfill_htlcs[0].clone(), commitment_signed.clone())) + }, + _ => panic!("Unexpected event"), + } + } + } + } + + macro_rules! last_update_fulfill_dance { + ($node: expr, $prev_node: expr) => { { $node.node.handle_update_fulfill_htlc(&$prev_node.node.get_our_node_id(), &next_msgs.as_ref().unwrap().0).unwrap(); - if $last_node { - check_added_monitors!($node, 0); + check_added_monitors!($node, 0); + assert!($node.node.get_and_clear_pending_msg_events().is_empty()); + commitment_signed_dance!($node, $prev_node, next_msgs.as_ref().unwrap().1, false); + } + } + } + macro_rules! mid_update_fulfill_dance { + ($node: expr, $prev_node: expr, $new_msgs: expr) => { + { + $node.node.handle_update_fulfill_htlc(&$prev_node.node.get_our_node_id(), &next_msgs.as_ref().unwrap().0).unwrap(); + check_added_monitors!($node, 1); + let new_next_msgs = if $new_msgs { + get_next_msgs!($node) } else { - check_added_monitors!($node, 1); - } + assert!($node.node.get_and_clear_pending_msg_events().is_empty()); + None + }; commitment_signed_dance!($node, $prev_node, next_msgs.as_ref().unwrap().1, false); + next_msgs = new_next_msgs; } } } - let mut expected_next_node = expected_route.last().unwrap().node.get_our_node_id(); let mut prev_node = expected_route.last().unwrap(); for (idx, node) in expected_route.iter().rev().enumerate() { assert_eq!(expected_next_node, node.node.get_our_node_id()); + let update_next_msgs = !skip_last || idx != expected_route.len() - 1; if next_msgs.is_some() { - update_fulfill_dance!(node, prev_node, false); - } - - let events = node.node.get_and_clear_pending_msg_events(); - if !skip_last || idx != expected_route.len() - 1 { - assert_eq!(events.len(), 1); - match events[0] { - MessageSendEvent::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 update_fee, 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()); - assert!(update_fee.is_none()); - expected_next_node = node_id.clone(); - next_msgs = Some((update_fulfill_htlcs[0].clone(), commitment_signed.clone())); - }, - _ => panic!("Unexpected event"), - } + mid_update_fulfill_dance!(node, prev_node, update_next_msgs); + } else if update_next_msgs { + next_msgs = get_next_msgs!(node); } else { - assert!(events.is_empty()); + assert!(node.node.get_and_clear_pending_msg_events().is_empty()); } if !skip_last && idx == expected_route.len() - 1 { assert_eq!(expected_next_node, origin_node.node.get_our_node_id()); @@ -3258,7 +3948,7 @@ mod tests { } if !skip_last { - update_fulfill_dance!(origin_node, expected_route.first().unwrap(), true); + last_update_fulfill_dance!(origin_node, expected_route.first().unwrap()); let events = origin_node.node.get_and_clear_pending_events(); assert_eq!(events.len(), 1); match events[0] { @@ -3392,10 +4082,13 @@ mod tests { let mut seed = [0; 32]; rng.fill_bytes(&mut seed); let keys_manager = Arc::new(keysinterface::KeysManager::new(&seed, Network::Testnet, Arc::clone(&logger))); - let chan_monitor = Arc::new(test_utils::TestChannelMonitor::new(chain_monitor.clone(), tx_broadcaster.clone())); - let node = ChannelManager::new(0, true, Network::Testnet, feeest.clone(), chan_monitor.clone(), chain_monitor.clone(), tx_broadcaster.clone(), Arc::clone(&logger), keys_manager.clone()).unwrap(); + let chan_monitor = Arc::new(test_utils::TestChannelMonitor::new(chain_monitor.clone(), tx_broadcaster.clone(), logger.clone())); + let mut config = UserConfig::new(); + config.channel_options.announced_channel = true; + config.channel_limits.force_announced_channel_preference = false; + let node = ChannelManager::new(Network::Testnet, feeest.clone(), chan_monitor.clone(), chain_monitor.clone(), tx_broadcaster.clone(), Arc::clone(&logger), keys_manager.clone(), config).unwrap(); let router = Router::new(PublicKey::from_secret_key(&secp_ctx, &keys_manager.get_node_secret()), chain_monitor.clone(), Arc::clone(&logger)); - nodes.push(Node { chain_monitor, tx_broadcaster, chan_monitor, node, router, + nodes.push(Node { chain_monitor, tx_broadcaster, chan_monitor, node, router, node_seed: seed, network_payment_count: payment_count.clone(), network_chan_count: chan_count.clone(), }); @@ -3468,40 +4161,45 @@ mod tests { // ...now when the messages get delivered everyone should be happy nodes[0].node.handle_update_add_htlc(&nodes[1].node.get_our_node_id(), &payment_event.msgs[0]).unwrap(); - let (as_revoke_msg, as_commitment_signed) = nodes[0].node.handle_commitment_signed(&nodes[1].node.get_our_node_id(), &payment_event.commitment_msg).unwrap(); // (2) - assert!(as_commitment_signed.is_none()); // nodes[0] is awaiting nodes[1] revoke_and_ack + nodes[0].node.handle_commitment_signed(&nodes[1].node.get_our_node_id(), &payment_event.commitment_msg).unwrap(); // (2) + let as_revoke_and_ack = get_event_msg!(nodes[0], MessageSendEvent::SendRevokeAndACK, nodes[1].node.get_our_node_id()); + // nodes[0] is awaiting nodes[1] revoke_and_ack so get_event_msg's assert(len == 1) passes check_added_monitors!(nodes[0], 1); // deliver(1), generate (3): - let (bs_revoke_msg, bs_commitment_signed) = nodes[1].node.handle_commitment_signed(&nodes[0].node.get_our_node_id(), commitment_signed).unwrap(); - assert!(bs_commitment_signed.is_none()); // nodes[1] is awaiting nodes[0] revoke_and_ack + nodes[1].node.handle_commitment_signed(&nodes[0].node.get_our_node_id(), commitment_signed).unwrap(); + let bs_revoke_and_ack = get_event_msg!(nodes[1], MessageSendEvent::SendRevokeAndACK, nodes[0].node.get_our_node_id()); + // nodes[1] is awaiting nodes[0] revoke_and_ack so get_event_msg's assert(len == 1) passes check_added_monitors!(nodes[1], 1); - let bs_update = nodes[1].node.handle_revoke_and_ack(&nodes[0].node.get_our_node_id(), &as_revoke_msg).unwrap(); // deliver (2) - assert!(bs_update.as_ref().unwrap().update_add_htlcs.is_empty()); // (4) - assert!(bs_update.as_ref().unwrap().update_fulfill_htlcs.is_empty()); // (4) - assert!(bs_update.as_ref().unwrap().update_fail_htlcs.is_empty()); // (4) - assert!(bs_update.as_ref().unwrap().update_fail_malformed_htlcs.is_empty()); // (4) - assert!(bs_update.as_ref().unwrap().update_fee.is_none()); // (4) + nodes[1].node.handle_revoke_and_ack(&nodes[0].node.get_our_node_id(), &as_revoke_and_ack).unwrap(); // deliver (2) + let bs_update = get_htlc_update_msgs!(nodes[1], nodes[0].node.get_our_node_id()); + assert!(bs_update.update_add_htlcs.is_empty()); // (4) + assert!(bs_update.update_fulfill_htlcs.is_empty()); // (4) + assert!(bs_update.update_fail_htlcs.is_empty()); // (4) + assert!(bs_update.update_fail_malformed_htlcs.is_empty()); // (4) + assert!(bs_update.update_fee.is_none()); // (4) check_added_monitors!(nodes[1], 1); - let as_update = nodes[0].node.handle_revoke_and_ack(&nodes[1].node.get_our_node_id(), &bs_revoke_msg).unwrap(); // deliver (3) - assert!(as_update.as_ref().unwrap().update_add_htlcs.is_empty()); // (5) - assert!(as_update.as_ref().unwrap().update_fulfill_htlcs.is_empty()); // (5) - assert!(as_update.as_ref().unwrap().update_fail_htlcs.is_empty()); // (5) - assert!(as_update.as_ref().unwrap().update_fail_malformed_htlcs.is_empty()); // (5) - assert!(as_update.as_ref().unwrap().update_fee.is_none()); // (5) + nodes[0].node.handle_revoke_and_ack(&nodes[1].node.get_our_node_id(), &bs_revoke_and_ack).unwrap(); // deliver (3) + let as_update = get_htlc_update_msgs!(nodes[0], nodes[1].node.get_our_node_id()); + assert!(as_update.update_add_htlcs.is_empty()); // (5) + assert!(as_update.update_fulfill_htlcs.is_empty()); // (5) + assert!(as_update.update_fail_htlcs.is_empty()); // (5) + assert!(as_update.update_fail_malformed_htlcs.is_empty()); // (5) + assert!(as_update.update_fee.is_none()); // (5) check_added_monitors!(nodes[0], 1); - let (as_second_revoke, as_second_commitment_signed) = nodes[0].node.handle_commitment_signed(&nodes[1].node.get_our_node_id(), &bs_update.unwrap().commitment_signed).unwrap(); // deliver (4) - assert!(as_second_commitment_signed.is_none()); // only (6) + nodes[0].node.handle_commitment_signed(&nodes[1].node.get_our_node_id(), &bs_update.commitment_signed).unwrap(); // deliver (4) + let as_second_revoke = get_event_msg!(nodes[0], MessageSendEvent::SendRevokeAndACK, nodes[1].node.get_our_node_id()); + // only (6) so get_event_msg's assert(len == 1) passes check_added_monitors!(nodes[0], 1); - let (bs_second_revoke, bs_second_commitment_signed) = nodes[1].node.handle_commitment_signed(&nodes[0].node.get_our_node_id(), &as_update.unwrap().commitment_signed).unwrap(); // deliver (5) - assert!(bs_second_commitment_signed.is_none()); + nodes[1].node.handle_commitment_signed(&nodes[0].node.get_our_node_id(), &as_update.commitment_signed).unwrap(); // deliver (5) + let bs_second_revoke = get_event_msg!(nodes[1], MessageSendEvent::SendRevokeAndACK, nodes[0].node.get_our_node_id()); check_added_monitors!(nodes[1], 1); - assert!(nodes[0].node.handle_revoke_and_ack(&nodes[1].node.get_our_node_id(), &bs_second_revoke).unwrap().is_none()); + nodes[0].node.handle_revoke_and_ack(&nodes[1].node.get_our_node_id(), &bs_second_revoke).unwrap(); check_added_monitors!(nodes[0], 1); let events_2 = nodes[0].node.get_and_clear_pending_events(); @@ -3511,7 +4209,7 @@ mod tests { _ => panic!("Unexpected event"), } - assert!(nodes[1].node.handle_revoke_and_ack(&nodes[0].node.get_our_node_id(), &as_second_revoke).unwrap().is_none()); // deliver (6) + nodes[1].node.handle_revoke_and_ack(&nodes[0].node.get_our_node_id(), &as_second_revoke).unwrap(); // deliver (6) check_added_monitors!(nodes[1], 1); } @@ -3564,11 +4262,12 @@ mod tests { // ...now when the messages get delivered everyone should be happy nodes[0].node.handle_update_add_htlc(&nodes[1].node.get_our_node_id(), &payment_event.msgs[0]).unwrap(); - let (as_revoke_msg, as_commitment_signed) = nodes[0].node.handle_commitment_signed(&nodes[1].node.get_our_node_id(), &payment_event.commitment_msg).unwrap(); // (2) - assert!(as_commitment_signed.is_none()); // nodes[0] is awaiting nodes[1] revoke_and_ack + nodes[0].node.handle_commitment_signed(&nodes[1].node.get_our_node_id(), &payment_event.commitment_msg).unwrap(); // (2) + let as_revoke_msg = get_event_msg!(nodes[0], MessageSendEvent::SendRevokeAndACK, nodes[1].node.get_our_node_id()); + // nodes[0] is awaiting nodes[1] revoke_and_ack so get_event_msg's assert(len == 1) passes check_added_monitors!(nodes[0], 1); - assert!(nodes[1].node.handle_revoke_and_ack(&nodes[0].node.get_our_node_id(), &as_revoke_msg).unwrap().is_none()); // deliver (2) + nodes[1].node.handle_revoke_and_ack(&nodes[0].node.get_our_node_id(), &as_revoke_msg).unwrap(); // deliver (2) check_added_monitors!(nodes[1], 1); // We can't continue, sadly, because our (1) now has a bogus signature @@ -3623,7 +4322,8 @@ mod tests { // Deliver first update_fee/commitment_signed pair, generating (1) and (2): nodes[1].node.handle_update_fee(&nodes[0].node.get_our_node_id(), update_msg_1).unwrap(); - let (bs_revoke_msg, bs_commitment_signed) = nodes[1].node.handle_commitment_signed(&nodes[0].node.get_our_node_id(), commitment_signed_1).unwrap(); + nodes[1].node.handle_commitment_signed(&nodes[0].node.get_our_node_id(), commitment_signed_1).unwrap(); + let (bs_revoke_msg, bs_commitment_signed) = get_revoke_commit_msgs!(nodes[1], nodes[0].node.get_our_node_id()); check_added_monitors!(nodes[1], 1); // nodes[0] is awaiting a revoke from nodes[1] before it will create a new commitment @@ -3645,36 +4345,43 @@ mod tests { nodes[1].node.handle_update_fee(&nodes[0].node.get_our_node_id(), &update_msg_2).unwrap(); // Deliver (1), generating (3) and (4) - let as_second_update = nodes[0].node.handle_revoke_and_ack(&nodes[1].node.get_our_node_id(), &bs_revoke_msg).unwrap(); + nodes[0].node.handle_revoke_and_ack(&nodes[1].node.get_our_node_id(), &bs_revoke_msg).unwrap(); + let as_second_update = get_htlc_update_msgs!(nodes[0], nodes[1].node.get_our_node_id()); check_added_monitors!(nodes[0], 1); - assert!(as_second_update.as_ref().unwrap().update_add_htlcs.is_empty()); - assert!(as_second_update.as_ref().unwrap().update_fulfill_htlcs.is_empty()); - assert!(as_second_update.as_ref().unwrap().update_fail_htlcs.is_empty()); - assert!(as_second_update.as_ref().unwrap().update_fail_malformed_htlcs.is_empty()); + assert!(as_second_update.update_add_htlcs.is_empty()); + assert!(as_second_update.update_fulfill_htlcs.is_empty()); + assert!(as_second_update.update_fail_htlcs.is_empty()); + assert!(as_second_update.update_fail_malformed_htlcs.is_empty()); // Check that the update_fee newly generated matches what we delivered: - assert_eq!(as_second_update.as_ref().unwrap().update_fee.as_ref().unwrap().channel_id, update_msg_2.channel_id); - assert_eq!(as_second_update.as_ref().unwrap().update_fee.as_ref().unwrap().feerate_per_kw, update_msg_2.feerate_per_kw); + assert_eq!(as_second_update.update_fee.as_ref().unwrap().channel_id, update_msg_2.channel_id); + assert_eq!(as_second_update.update_fee.as_ref().unwrap().feerate_per_kw, update_msg_2.feerate_per_kw); // Deliver (2) commitment_signed - let (as_revoke_msg, as_commitment_signed) = nodes[0].node.handle_commitment_signed(&nodes[1].node.get_our_node_id(), bs_commitment_signed.as_ref().unwrap()).unwrap(); + nodes[0].node.handle_commitment_signed(&nodes[1].node.get_our_node_id(), &bs_commitment_signed).unwrap(); + let as_revoke_msg = get_event_msg!(nodes[0], MessageSendEvent::SendRevokeAndACK, nodes[1].node.get_our_node_id()); check_added_monitors!(nodes[0], 1); - assert!(as_commitment_signed.is_none()); + // No commitment_signed so get_event_msg's assert(len == 1) passes - assert!(nodes[1].node.handle_revoke_and_ack(&nodes[0].node.get_our_node_id(), &as_revoke_msg).unwrap().is_none()); + nodes[1].node.handle_revoke_and_ack(&nodes[0].node.get_our_node_id(), &as_revoke_msg).unwrap(); + assert!(nodes[1].node.get_and_clear_pending_msg_events().is_empty()); check_added_monitors!(nodes[1], 1); // Delever (4) - let (bs_second_revoke, bs_second_commitment) = nodes[1].node.handle_commitment_signed(&nodes[0].node.get_our_node_id(), &as_second_update.unwrap().commitment_signed).unwrap(); + nodes[1].node.handle_commitment_signed(&nodes[0].node.get_our_node_id(), &as_second_update.commitment_signed).unwrap(); + let (bs_second_revoke, bs_second_commitment) = get_revoke_commit_msgs!(nodes[1], nodes[0].node.get_our_node_id()); check_added_monitors!(nodes[1], 1); - assert!(nodes[0].node.handle_revoke_and_ack(&nodes[1].node.get_our_node_id(), &bs_second_revoke).unwrap().is_none()); + nodes[0].node.handle_revoke_and_ack(&nodes[1].node.get_our_node_id(), &bs_second_revoke).unwrap(); + assert!(nodes[0].node.get_and_clear_pending_msg_events().is_empty()); check_added_monitors!(nodes[0], 1); - let (as_second_revoke, as_second_commitment) = nodes[0].node.handle_commitment_signed(&nodes[1].node.get_our_node_id(), &bs_second_commitment.unwrap()).unwrap(); - assert!(as_second_commitment.is_none()); + nodes[0].node.handle_commitment_signed(&nodes[1].node.get_our_node_id(), &bs_second_commitment).unwrap(); + let as_second_revoke = get_event_msg!(nodes[0], MessageSendEvent::SendRevokeAndACK, nodes[1].node.get_our_node_id()); + // No commitment_signed so get_event_msg's assert(len == 1) passes check_added_monitors!(nodes[0], 1); - assert!(nodes[1].node.handle_revoke_and_ack(&nodes[0].node.get_our_node_id(), &as_second_revoke).unwrap().is_none()); + nodes[1].node.handle_revoke_and_ack(&nodes[0].node.get_our_node_id(), &as_second_revoke).unwrap(); + assert!(nodes[1].node.get_and_clear_pending_msg_events().is_empty()); check_added_monitors!(nodes[1], 1); } @@ -3694,6 +4401,7 @@ mod tests { let feerate = get_feerate!(nodes[0]); nodes[0].node.update_fee(channel_id, feerate+20).unwrap(); + check_added_monitors!(nodes[0], 1); let events_0 = nodes[0].node.get_and_clear_pending_msg_events(); assert_eq!(events_0.len(), 1); @@ -3705,21 +4413,21 @@ mod tests { }; nodes[1].node.handle_update_fee(&nodes[0].node.get_our_node_id(), update_msg.unwrap()).unwrap(); - let (revoke_msg, commitment_signed) = nodes[1].node.handle_commitment_signed(&nodes[0].node.get_our_node_id(), commitment_signed).unwrap(); - let commitment_signed = commitment_signed.unwrap(); - check_added_monitors!(nodes[0], 1); + nodes[1].node.handle_commitment_signed(&nodes[0].node.get_our_node_id(), commitment_signed).unwrap(); + let (revoke_msg, commitment_signed) = get_revoke_commit_msgs!(nodes[1], nodes[0].node.get_our_node_id()); check_added_monitors!(nodes[1], 1); - let resp_option = nodes[0].node.handle_revoke_and_ack(&nodes[1].node.get_our_node_id(), &revoke_msg).unwrap(); - assert!(resp_option.is_none()); + nodes[0].node.handle_revoke_and_ack(&nodes[1].node.get_our_node_id(), &revoke_msg).unwrap(); + assert!(nodes[0].node.get_and_clear_pending_msg_events().is_empty()); check_added_monitors!(nodes[0], 1); - let (revoke_msg, commitment_signed) = nodes[0].node.handle_commitment_signed(&nodes[1].node.get_our_node_id(), &commitment_signed).unwrap(); - assert!(commitment_signed.is_none()); + nodes[0].node.handle_commitment_signed(&nodes[1].node.get_our_node_id(), &commitment_signed).unwrap(); + let revoke_msg = get_event_msg!(nodes[0], MessageSendEvent::SendRevokeAndACK, nodes[1].node.get_our_node_id()); + // No commitment_signed so get_event_msg's assert(len == 1) passes check_added_monitors!(nodes[0], 1); - let resp_option = nodes[1].node.handle_revoke_and_ack(&nodes[0].node.get_our_node_id(), &revoke_msg).unwrap(); - assert!(resp_option.is_none()); + nodes[1].node.handle_revoke_and_ack(&nodes[0].node.get_our_node_id(), &revoke_msg).unwrap(); + assert!(nodes[1].node.get_and_clear_pending_msg_events().is_empty()); check_added_monitors!(nodes[1], 1); } @@ -3742,6 +4450,7 @@ mod tests { let feerate = get_feerate!(nodes[0]); nodes[0].node.update_fee(channel_id, feerate+20).unwrap(); + check_added_monitors!(nodes[0], 1); let events_0 = nodes[0].node.get_and_clear_pending_msg_events(); assert_eq!(events_0.len(), 1); @@ -3752,9 +4461,8 @@ mod tests { _ => panic!("Unexpected event"), }; nodes[1].node.handle_update_fee(&nodes[0].node.get_our_node_id(), update_msg.unwrap()).unwrap(); - check_added_monitors!(nodes[0], 1); - let (revoke_msg, commitment_signed) = nodes[1].node.handle_commitment_signed(&nodes[0].node.get_our_node_id(), commitment_signed).unwrap(); - let commitment_signed = commitment_signed.unwrap(); + nodes[1].node.handle_commitment_signed(&nodes[0].node.get_our_node_id(), commitment_signed).unwrap(); + let (revoke_msg, commitment_signed) = get_revoke_commit_msgs!(nodes[1], nodes[0].node.get_our_node_id()); check_added_monitors!(nodes[1], 1); let route = nodes[1].router.get_route(&nodes[0].node.get_our_node_id(), None, &Vec::new(), 800000, TEST_FINAL_CLTV).unwrap(); @@ -3772,17 +4480,19 @@ mod tests { assert!(nodes[0].node.get_and_clear_pending_msg_events().is_empty()); // node[1] has nothing to do - let resp_option = nodes[0].node.handle_revoke_and_ack(&nodes[1].node.get_our_node_id(), &revoke_msg).unwrap(); - assert!(resp_option.is_none()); + nodes[0].node.handle_revoke_and_ack(&nodes[1].node.get_our_node_id(), &revoke_msg).unwrap(); + assert!(nodes[0].node.get_and_clear_pending_msg_events().is_empty()); check_added_monitors!(nodes[0], 1); - let (revoke_msg, commitment_signed) = nodes[0].node.handle_commitment_signed(&nodes[1].node.get_our_node_id(), &commitment_signed).unwrap(); - assert!(commitment_signed.is_none()); + nodes[0].node.handle_commitment_signed(&nodes[1].node.get_our_node_id(), &commitment_signed).unwrap(); + let revoke_msg = get_event_msg!(nodes[0], MessageSendEvent::SendRevokeAndACK, nodes[1].node.get_our_node_id()); + // No commitment_signed so get_event_msg's assert(len == 1) passes check_added_monitors!(nodes[0], 1); - let resp_option = nodes[1].node.handle_revoke_and_ack(&nodes[0].node.get_our_node_id(), &revoke_msg).unwrap(); + nodes[1].node.handle_revoke_and_ack(&nodes[0].node.get_our_node_id(), &revoke_msg).unwrap(); + check_added_monitors!(nodes[1], 1); // AwaitingRemoteRevoke ends here - let commitment_update = resp_option.unwrap(); + let commitment_update = get_htlc_update_msgs!(nodes[1], nodes[0].node.get_our_node_id()); assert_eq!(commitment_update.update_add_htlcs.len(), 1); assert_eq!(commitment_update.update_fulfill_htlcs.len(), 0); assert_eq!(commitment_update.update_fail_htlcs.len(), 0); @@ -3790,20 +4500,22 @@ mod tests { assert_eq!(commitment_update.update_fee.is_none(), true); nodes[0].node.handle_update_add_htlc(&nodes[1].node.get_our_node_id(), &commitment_update.update_add_htlcs[0]).unwrap(); - let (revoke, commitment_signed) = nodes[0].node.handle_commitment_signed(&nodes[1].node.get_our_node_id(), &commitment_update.commitment_signed).unwrap(); + nodes[0].node.handle_commitment_signed(&nodes[1].node.get_our_node_id(), &commitment_update.commitment_signed).unwrap(); check_added_monitors!(nodes[0], 1); + let (revoke, commitment_signed) = get_revoke_commit_msgs!(nodes[0], nodes[1].node.get_our_node_id()); + + nodes[1].node.handle_revoke_and_ack(&nodes[0].node.get_our_node_id(), &revoke).unwrap(); check_added_monitors!(nodes[1], 1); - let commitment_signed = commitment_signed.unwrap(); - let resp_option = nodes[1].node.handle_revoke_and_ack(&nodes[0].node.get_our_node_id(), &revoke).unwrap(); - check_added_monitors!(nodes[1], 1); - assert!(resp_option.is_none()); + assert!(nodes[1].node.get_and_clear_pending_msg_events().is_empty()); - let (revoke, commitment_signed) = nodes[1].node.handle_commitment_signed(&nodes[0].node.get_our_node_id(), &commitment_signed).unwrap(); + nodes[1].node.handle_commitment_signed(&nodes[0].node.get_our_node_id(), &commitment_signed).unwrap(); check_added_monitors!(nodes[1], 1); - assert!(commitment_signed.is_none()); - let resp_option = nodes[0].node.handle_revoke_and_ack(&nodes[1].node.get_our_node_id(), &revoke).unwrap(); + let revoke = get_event_msg!(nodes[1], MessageSendEvent::SendRevokeAndACK, nodes[0].node.get_our_node_id()); + // No commitment_signed so get_event_msg's assert(len == 1) passes + + nodes[0].node.handle_revoke_and_ack(&nodes[1].node.get_our_node_id(), &revoke).unwrap(); check_added_monitors!(nodes[0], 1); - assert!(resp_option.is_none()); + assert!(nodes[0].node.get_and_clear_pending_msg_events().is_empty()); let events = nodes[0].node.get_and_clear_pending_events(); assert_eq!(events.len(), 1); @@ -3859,6 +4571,7 @@ mod tests { // Create and deliver (1)... let feerate = get_feerate!(nodes[0]); nodes[0].node.update_fee(channel_id, feerate+20).unwrap(); + check_added_monitors!(nodes[0], 1); let events_0 = nodes[0].node.get_and_clear_pending_msg_events(); assert_eq!(events_0.len(), 1); @@ -3871,18 +4584,18 @@ mod tests { nodes[1].node.handle_update_fee(&nodes[0].node.get_our_node_id(), update_msg.unwrap()).unwrap(); // Generate (2) and (3): - let (revoke_msg, commitment_signed) = nodes[1].node.handle_commitment_signed(&nodes[0].node.get_our_node_id(), commitment_signed).unwrap(); - let commitment_signed_0 = commitment_signed.unwrap(); - check_added_monitors!(nodes[0], 1); + nodes[1].node.handle_commitment_signed(&nodes[0].node.get_our_node_id(), commitment_signed).unwrap(); + let (revoke_msg, commitment_signed_0) = get_revoke_commit_msgs!(nodes[1], nodes[0].node.get_our_node_id()); check_added_monitors!(nodes[1], 1); // Deliver (2): - let resp_option = nodes[0].node.handle_revoke_and_ack(&nodes[1].node.get_our_node_id(), &revoke_msg).unwrap(); - assert!(resp_option.is_none()); + nodes[0].node.handle_revoke_and_ack(&nodes[1].node.get_our_node_id(), &revoke_msg).unwrap(); + assert!(nodes[0].node.get_and_clear_pending_msg_events().is_empty()); check_added_monitors!(nodes[0], 1); // Create and deliver (4)... nodes[0].node.update_fee(channel_id, feerate+30).unwrap(); + check_added_monitors!(nodes[0], 1); let events_0 = nodes[0].node.get_and_clear_pending_msg_events(); assert_eq!(events_0.len(), 1); let (update_msg, commitment_signed) = match events_0[0] { @@ -3891,36 +4604,44 @@ mod tests { }, _ => panic!("Unexpected event"), }; - nodes[1].node.handle_update_fee(&nodes[0].node.get_our_node_id(), update_msg.unwrap()).unwrap(); - let (revoke_msg, commitment_signed) = nodes[1].node.handle_commitment_signed(&nodes[0].node.get_our_node_id(), commitment_signed).unwrap(); - // ... creating (5) - assert!(commitment_signed.is_none()); - check_added_monitors!(nodes[0], 1); + nodes[1].node.handle_update_fee(&nodes[0].node.get_our_node_id(), update_msg.unwrap()).unwrap(); + nodes[1].node.handle_commitment_signed(&nodes[0].node.get_our_node_id(), commitment_signed).unwrap(); check_added_monitors!(nodes[1], 1); + // ... creating (5) + let revoke_msg = get_event_msg!(nodes[1], MessageSendEvent::SendRevokeAndACK, nodes[0].node.get_our_node_id()); + // No commitment_signed so get_event_msg's assert(len == 1) passes // Handle (3), creating (6): - let (revoke_msg_0, commitment_signed) = nodes[0].node.handle_commitment_signed(&nodes[1].node.get_our_node_id(), &commitment_signed_0).unwrap(); - assert!(commitment_signed.is_none()); + nodes[0].node.handle_commitment_signed(&nodes[1].node.get_our_node_id(), &commitment_signed_0).unwrap(); check_added_monitors!(nodes[0], 1); + let revoke_msg_0 = get_event_msg!(nodes[0], MessageSendEvent::SendRevokeAndACK, nodes[1].node.get_our_node_id()); + // No commitment_signed so get_event_msg's assert(len == 1) passes // Deliver (5): - let resp_option = nodes[0].node.handle_revoke_and_ack(&nodes[1].node.get_our_node_id(), &revoke_msg).unwrap(); - assert!(resp_option.is_none()); + nodes[0].node.handle_revoke_and_ack(&nodes[1].node.get_our_node_id(), &revoke_msg).unwrap(); + assert!(nodes[0].node.get_and_clear_pending_msg_events().is_empty()); check_added_monitors!(nodes[0], 1); // Deliver (6), creating (7): - let resp_option = nodes[1].node.handle_revoke_and_ack(&nodes[0].node.get_our_node_id(), &revoke_msg_0).unwrap(); - let commitment_signed = resp_option.unwrap().commitment_signed; + nodes[1].node.handle_revoke_and_ack(&nodes[0].node.get_our_node_id(), &revoke_msg_0).unwrap(); + let commitment_update = get_htlc_update_msgs!(nodes[1], nodes[0].node.get_our_node_id()); + assert!(commitment_update.update_add_htlcs.is_empty()); + assert!(commitment_update.update_fulfill_htlcs.is_empty()); + assert!(commitment_update.update_fail_htlcs.is_empty()); + assert!(commitment_update.update_fail_malformed_htlcs.is_empty()); + assert!(commitment_update.update_fee.is_none()); check_added_monitors!(nodes[1], 1); // Deliver (7) - let (revoke_msg, commitment_signed) = nodes[0].node.handle_commitment_signed(&nodes[1].node.get_our_node_id(), &commitment_signed).unwrap(); - assert!(commitment_signed.is_none()); + nodes[0].node.handle_commitment_signed(&nodes[1].node.get_our_node_id(), &commitment_update.commitment_signed).unwrap(); check_added_monitors!(nodes[0], 1); - let resp_option = nodes[1].node.handle_revoke_and_ack(&nodes[0].node.get_our_node_id(), &revoke_msg).unwrap(); - assert!(resp_option.is_none()); + let revoke_msg = get_event_msg!(nodes[0], MessageSendEvent::SendRevokeAndACK, nodes[1].node.get_our_node_id()); + // No commitment_signed so get_event_msg's assert(len == 1) passes + + nodes[1].node.handle_revoke_and_ack(&nodes[0].node.get_our_node_id(), &revoke_msg).unwrap(); check_added_monitors!(nodes[1], 1); + assert!(nodes[1].node.get_and_clear_pending_msg_events().is_empty()); assert_eq!(get_feerate!(nodes[0]), feerate + 30); assert_eq!(get_feerate!(nodes[1]), feerate + 30); @@ -4391,15 +5112,21 @@ mod tests { assert!(nodes[0].node.get_and_clear_pending_msg_events().is_empty()); // flush the pending htlc - let (as_revoke_and_ack, as_commitment_signed) = nodes[1].node.handle_commitment_signed(&nodes[0].node.get_our_node_id(), &payment_event_1.commitment_msg).unwrap(); + nodes[1].node.handle_commitment_signed(&nodes[0].node.get_our_node_id(), &payment_event_1.commitment_msg).unwrap(); + let (as_revoke_and_ack, as_commitment_signed) = get_revoke_commit_msgs!(nodes[1], nodes[0].node.get_our_node_id()); check_added_monitors!(nodes[1], 1); - let commitment_update_2 = nodes[0].node.handle_revoke_and_ack(&nodes[1].node.get_our_node_id(), &as_revoke_and_ack).unwrap().unwrap(); + nodes[0].node.handle_revoke_and_ack(&nodes[1].node.get_our_node_id(), &as_revoke_and_ack).unwrap(); check_added_monitors!(nodes[0], 1); - let (bs_revoke_and_ack, bs_none) = nodes[0].node.handle_commitment_signed(&nodes[1].node.get_our_node_id(), &as_commitment_signed.unwrap()).unwrap(); - assert!(bs_none.is_none()); + let commitment_update_2 = get_htlc_update_msgs!(nodes[0], nodes[1].node.get_our_node_id()); + + nodes[0].node.handle_commitment_signed(&nodes[1].node.get_our_node_id(), &as_commitment_signed).unwrap(); + let bs_revoke_and_ack = get_event_msg!(nodes[0], MessageSendEvent::SendRevokeAndACK, nodes[1].node.get_our_node_id()); + // No commitment_signed so get_event_msg's assert(len == 1) passes check_added_monitors!(nodes[0], 1); - assert!(nodes[1].node.handle_revoke_and_ack(&nodes[0].node.get_our_node_id(), &bs_revoke_and_ack).unwrap().is_none()); + + nodes[1].node.handle_revoke_and_ack(&nodes[0].node.get_our_node_id(), &bs_revoke_and_ack).unwrap(); + assert!(nodes[1].node.get_and_clear_pending_msg_events().is_empty()); check_added_monitors!(nodes[1], 1); expect_pending_htlcs_forwardable!(nodes[1]); @@ -4875,6 +5602,7 @@ mod tests { nodes[2].node.handle_update_add_htlc(&nodes[1].node.get_our_node_id(), &payment_event.msgs[0]).unwrap(); nodes[2].node.handle_commitment_signed(&nodes[1].node.get_our_node_id(), &payment_event.commitment_msg).unwrap(); check_added_monitors!(nodes[2], 1); + let (_, _) = get_revoke_commit_msgs!(nodes[2], nodes[1].node.get_our_node_id()); // nodes[2] now has the latest commitment transaction, but hasn't revoked its previous // state or updated nodes[1]' state. Now force-close and broadcast that commitment/HTLC @@ -4965,15 +5693,90 @@ mod tests { assert_eq!(channel_state.short_to_id.len(), 0); } + macro_rules! get_chan_reestablish_msgs { + ($src_node: expr, $dst_node: expr) => { + { + let mut res = Vec::with_capacity(1); + for msg in $src_node.node.get_and_clear_pending_msg_events() { + if let MessageSendEvent::SendChannelReestablish { ref node_id, ref msg } = msg { + assert_eq!(*node_id, $dst_node.node.get_our_node_id()); + res.push(msg.clone()); + } else { + panic!("Unexpected event") + } + } + res + } + } + } + + macro_rules! handle_chan_reestablish_msgs { + ($src_node: expr, $dst_node: expr) => { + { + let msg_events = $src_node.node.get_and_clear_pending_msg_events(); + let mut idx = 0; + let funding_locked = if let Some(&MessageSendEvent::SendFundingLocked { ref node_id, ref msg }) = msg_events.get(0) { + idx += 1; + assert_eq!(*node_id, $dst_node.node.get_our_node_id()); + Some(msg.clone()) + } else { + None + }; + + let mut revoke_and_ack = None; + let mut commitment_update = None; + let order = if let Some(ev) = msg_events.get(idx) { + idx += 1; + match ev { + &MessageSendEvent::SendRevokeAndACK { ref node_id, ref msg } => { + assert_eq!(*node_id, $dst_node.node.get_our_node_id()); + revoke_and_ack = Some(msg.clone()); + RAACommitmentOrder::RevokeAndACKFirst + }, + &MessageSendEvent::UpdateHTLCs { ref node_id, ref updates } => { + assert_eq!(*node_id, $dst_node.node.get_our_node_id()); + commitment_update = Some(updates.clone()); + RAACommitmentOrder::CommitmentFirst + }, + _ => panic!("Unexpected event"), + } + } else { + RAACommitmentOrder::CommitmentFirst + }; + + if let Some(ev) = msg_events.get(idx) { + match ev { + &MessageSendEvent::SendRevokeAndACK { ref node_id, ref msg } => { + assert_eq!(*node_id, $dst_node.node.get_our_node_id()); + assert!(revoke_and_ack.is_none()); + revoke_and_ack = Some(msg.clone()); + }, + &MessageSendEvent::UpdateHTLCs { ref node_id, ref updates } => { + assert_eq!(*node_id, $dst_node.node.get_our_node_id()); + assert!(commitment_update.is_none()); + commitment_update = Some(updates.clone()); + }, + _ => panic!("Unexpected event"), + } + } + + (funding_locked, revoke_and_ack, commitment_update, order) + } + } + } + /// pending_htlc_adds includes both the holding cell and in-flight update_add_htlcs, whereas /// for claims/fails they are separated out. fn reconnect_nodes(node_a: &Node, node_b: &Node, pre_all_htlcs: bool, pending_htlc_adds: (i64, i64), pending_htlc_claims: (usize, usize), pending_cell_htlc_claims: (usize, usize), pending_cell_htlc_fails: (usize, usize), pending_raa: (bool, bool)) { - 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()); + node_a.node.peer_connected(&node_b.node.get_our_node_id()); + let reestablish_1 = get_chan_reestablish_msgs!(node_a, node_b); + node_b.node.peer_connected(&node_a.node.get_our_node_id()); + let reestablish_2 = get_chan_reestablish_msgs!(node_b, node_a); 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()); + node_b.node.handle_channel_reestablish(&node_a.node.get_our_node_id(), &msg).unwrap(); + resp_1.push(handle_chan_reestablish_msgs!(node_b, node_a)); } if pending_cell_htlc_claims.0 != 0 || pending_cell_htlc_fails.0 != 0 { check_added_monitors!(node_b, 1); @@ -4983,7 +5786,8 @@ mod tests { 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()); + node_a.node.handle_channel_reestablish(&node_b.node.get_our_node_id(), &msg).unwrap(); + resp_2.push(handle_chan_reestablish_msgs!(node_a, node_b)); } if pending_cell_htlc_claims.1 != 0 || pending_cell_htlc_fails.1 != 0 { check_added_monitors!(node_a, 1); @@ -4997,15 +5801,21 @@ mod tests { for chan_msgs in resp_1.drain(..) { if pre_all_htlcs { - let a = node_a.node.handle_funding_locked(&node_b.node.get_our_node_id(), &chan_msgs.0.unwrap()); - let _announcement_sigs_opt = a.unwrap(); - //TODO: Test announcement_sigs re-sending when we've implemented it + node_a.node.handle_funding_locked(&node_b.node.get_our_node_id(), &chan_msgs.0.unwrap()).unwrap(); + let announcement_event = node_a.node.get_and_clear_pending_msg_events(); + if !announcement_event.is_empty() { + assert_eq!(announcement_event.len(), 1); + if let MessageSendEvent::SendAnnouncementSignatures { .. } = announcement_event[0] { + //TODO: Test announcement_sigs re-sending + } else { panic!("Unexpected event!"); } + } } else { assert!(chan_msgs.0.is_none()); } if pending_raa.0 { - assert!(chan_msgs.3 == msgs::RAACommitmentOrder::RevokeAndACKFirst); - assert!(node_a.node.handle_revoke_and_ack(&node_b.node.get_our_node_id(), &chan_msgs.1.unwrap()).unwrap().is_none()); + assert!(chan_msgs.3 == RAACommitmentOrder::RevokeAndACKFirst); + node_a.node.handle_revoke_and_ack(&node_b.node.get_our_node_id(), &chan_msgs.1.unwrap()).unwrap(); + assert!(node_a.node.get_and_clear_pending_msg_events().is_empty()); check_added_monitors!(node_a, 1); } else { assert!(chan_msgs.1.is_none()); @@ -5033,10 +5843,12 @@ mod tests { if pending_htlc_adds.0 != -1 { // We use -1 to denote a response commitment_signed commitment_signed_dance!(node_a, node_b, commitment_update.commitment_signed, false); } else { - let (as_revoke_and_ack, as_commitment_signed) = node_a.node.handle_commitment_signed(&node_b.node.get_our_node_id(), &commitment_update.commitment_signed).unwrap(); + node_a.node.handle_commitment_signed(&node_b.node.get_our_node_id(), &commitment_update.commitment_signed).unwrap(); check_added_monitors!(node_a, 1); - assert!(as_commitment_signed.is_none()); - assert!(node_b.node.handle_revoke_and_ack(&node_a.node.get_our_node_id(), &as_revoke_and_ack).unwrap().is_none()); + let as_revoke_and_ack = get_event_msg!(node_a, MessageSendEvent::SendRevokeAndACK, node_b.node.get_our_node_id()); + // No commitment_signed so get_event_msg's assert(len == 1) passes + node_b.node.handle_revoke_and_ack(&node_a.node.get_our_node_id(), &as_revoke_and_ack).unwrap(); + assert!(node_b.node.get_and_clear_pending_msg_events().is_empty()); check_added_monitors!(node_b, 1); } } else { @@ -5046,14 +5858,21 @@ mod tests { 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 + node_b.node.handle_funding_locked(&node_a.node.get_our_node_id(), &chan_msgs.0.unwrap()).unwrap(); + let announcement_event = node_b.node.get_and_clear_pending_msg_events(); + if !announcement_event.is_empty() { + assert_eq!(announcement_event.len(), 1); + if let MessageSendEvent::SendAnnouncementSignatures { .. } = announcement_event[0] { + //TODO: Test announcement_sigs re-sending + } else { panic!("Unexpected event!"); } + } } else { assert!(chan_msgs.0.is_none()); } if pending_raa.1 { - assert!(chan_msgs.3 == msgs::RAACommitmentOrder::RevokeAndACKFirst); - assert!(node_b.node.handle_revoke_and_ack(&node_a.node.get_our_node_id(), &chan_msgs.1.unwrap()).unwrap().is_none()); + assert!(chan_msgs.3 == RAACommitmentOrder::RevokeAndACKFirst); + node_b.node.handle_revoke_and_ack(&node_a.node.get_our_node_id(), &chan_msgs.1.unwrap()).unwrap(); + assert!(node_b.node.get_and_clear_pending_msg_events().is_empty()); check_added_monitors!(node_b, 1); } else { assert!(chan_msgs.1.is_none()); @@ -5079,10 +5898,12 @@ mod tests { if pending_htlc_adds.1 != -1 { // We use -1 to denote a response commitment_signed commitment_signed_dance!(node_b, node_a, commitment_update.commitment_signed, false); } else { - let (bs_revoke_and_ack, bs_commitment_signed) = node_b.node.handle_commitment_signed(&node_a.node.get_our_node_id(), &commitment_update.commitment_signed).unwrap(); + node_b.node.handle_commitment_signed(&node_a.node.get_our_node_id(), &commitment_update.commitment_signed).unwrap(); check_added_monitors!(node_b, 1); - assert!(bs_commitment_signed.is_none()); - assert!(node_a.node.handle_revoke_and_ack(&node_b.node.get_our_node_id(), &bs_revoke_and_ack).unwrap().is_none()); + let bs_revoke_and_ack = get_event_msg!(node_b, MessageSendEvent::SendRevokeAndACK, node_a.node.get_our_node_id()); + // No commitment_signed so get_event_msg's assert(len == 1) passes + node_a.node.handle_revoke_and_ack(&node_b.node.get_our_node_id(), &bs_revoke_and_ack).unwrap(); + assert!(node_a.node.get_and_clear_pending_msg_events().is_empty()); check_added_monitors!(node_a, 1); } } else { @@ -5172,21 +5993,27 @@ mod tests { // Drop the payment_event messages, and let them get re-generated in reconnect_nodes! } else { nodes[1].node.handle_update_add_htlc(&nodes[0].node.get_our_node_id(), &payment_event.msgs[0]).unwrap(); - let (bs_revoke_and_ack, bs_commitment_signed) = nodes[1].node.handle_commitment_signed(&nodes[0].node.get_our_node_id(), &payment_event.commitment_msg).unwrap(); - check_added_monitors!(nodes[1], 1); - if messages_delivered >= 3 { - assert!(nodes[0].node.handle_revoke_and_ack(&nodes[1].node.get_our_node_id(), &bs_revoke_and_ack).unwrap().is_none()); - check_added_monitors!(nodes[0], 1); + nodes[1].node.handle_commitment_signed(&nodes[0].node.get_our_node_id(), &payment_event.commitment_msg).unwrap(); + check_added_monitors!(nodes[1], 1); + let (bs_revoke_and_ack, bs_commitment_signed) = get_revoke_commit_msgs!(nodes[1], nodes[0].node.get_our_node_id()); if messages_delivered >= 4 { - let (as_revoke_and_ack, as_commitment_signed) = nodes[0].node.handle_commitment_signed(&nodes[1].node.get_our_node_id(), &bs_commitment_signed.unwrap()).unwrap(); - assert!(as_commitment_signed.is_none()); + nodes[0].node.handle_revoke_and_ack(&nodes[1].node.get_our_node_id(), &bs_revoke_and_ack).unwrap(); + assert!(nodes[0].node.get_and_clear_pending_msg_events().is_empty()); check_added_monitors!(nodes[0], 1); if messages_delivered >= 5 { - assert!(nodes[1].node.handle_revoke_and_ack(&nodes[0].node.get_our_node_id(), &as_revoke_and_ack).unwrap().is_none()); - check_added_monitors!(nodes[1], 1); + nodes[0].node.handle_commitment_signed(&nodes[1].node.get_our_node_id(), &bs_commitment_signed).unwrap(); + let as_revoke_and_ack = get_event_msg!(nodes[0], MessageSendEvent::SendRevokeAndACK, nodes[1].node.get_our_node_id()); + // No commitment_signed so get_event_msg's assert(len == 1) passes + check_added_monitors!(nodes[0], 1); + + if messages_delivered >= 6 { + nodes[1].node.handle_revoke_and_ack(&nodes[0].node.get_our_node_id(), &as_revoke_and_ack).unwrap(); + assert!(nodes[1].node.get_and_clear_pending_msg_events().is_empty()); + check_added_monitors!(nodes[1], 1); + } } } } @@ -5194,20 +6021,20 @@ mod tests { 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); - if messages_delivered < 2 { + if messages_delivered < 3 { // Even if the funding_locked messages get exchanged, as long as nothing further was // received on either side, both sides will need to resend them. reconnect_nodes(&nodes[0], &nodes[1], true, (0, 1), (0, 0), (0, 0), (0, 0), (false, false)); - } else if messages_delivered == 2 { + } else if messages_delivered == 3 { // nodes[0] still wants its RAA + commitment_signed reconnect_nodes(&nodes[0], &nodes[1], false, (-1, 0), (0, 0), (0, 0), (0, 0), (true, false)); - } else if messages_delivered == 3 { + } else if messages_delivered == 4 { // nodes[0] still wants its commitment_signed reconnect_nodes(&nodes[0], &nodes[1], false, (-1, 0), (0, 0), (0, 0), (0, 0), (false, false)); - } else if messages_delivered == 4 { + } else if messages_delivered == 5 { // nodes[1] still wants its final RAA reconnect_nodes(&nodes[0], &nodes[1], false, (0, 0), (0, 0), (0, 0), (0, 0), (false, true)); - } else if messages_delivered == 5 { + } else if messages_delivered == 6 { // Everything was delivered... reconnect_nodes(&nodes[0], &nodes[1], false, (0, 0), (0, 0), (0, 0), (0, 0), (false, false)); } @@ -5267,20 +6094,24 @@ mod tests { } if messages_delivered >= 2 { - let (as_revoke_and_ack, as_commitment_signed) = nodes[0].node.handle_commitment_signed(&nodes[1].node.get_our_node_id(), &commitment_signed).unwrap(); + nodes[0].node.handle_commitment_signed(&nodes[1].node.get_our_node_id(), &commitment_signed).unwrap(); check_added_monitors!(nodes[0], 1); + let (as_revoke_and_ack, as_commitment_signed) = get_revoke_commit_msgs!(nodes[0], nodes[1].node.get_our_node_id()); if messages_delivered >= 3 { - assert!(nodes[1].node.handle_revoke_and_ack(&nodes[0].node.get_our_node_id(), &as_revoke_and_ack).unwrap().is_none()); + nodes[1].node.handle_revoke_and_ack(&nodes[0].node.get_our_node_id(), &as_revoke_and_ack).unwrap(); + assert!(nodes[1].node.get_and_clear_pending_msg_events().is_empty()); check_added_monitors!(nodes[1], 1); if messages_delivered >= 4 { - let (bs_revoke_and_ack, bs_commitment_signed) = nodes[1].node.handle_commitment_signed(&nodes[0].node.get_our_node_id(), &as_commitment_signed.unwrap()).unwrap(); - assert!(bs_commitment_signed.is_none()); + nodes[1].node.handle_commitment_signed(&nodes[0].node.get_our_node_id(), &as_commitment_signed).unwrap(); + let bs_revoke_and_ack = get_event_msg!(nodes[1], MessageSendEvent::SendRevokeAndACK, nodes[0].node.get_our_node_id()); + // No commitment_signed so get_event_msg's assert(len == 1) passes check_added_monitors!(nodes[1], 1); if messages_delivered >= 5 { - assert!(nodes[0].node.handle_revoke_and_ack(&nodes[1].node.get_our_node_id(), &bs_revoke_and_ack).unwrap().is_none()); + nodes[0].node.handle_revoke_and_ack(&nodes[1].node.get_our_node_id(), &bs_revoke_and_ack).unwrap(); + assert!(nodes[0].node.get_and_clear_pending_msg_events().is_empty()); check_added_monitors!(nodes[0], 1); } } @@ -5331,13 +6162,14 @@ mod tests { do_test_drop_messages_peer_disconnect(0); do_test_drop_messages_peer_disconnect(1); do_test_drop_messages_peer_disconnect(2); + do_test_drop_messages_peer_disconnect(3); } #[test] fn test_drop_messages_peer_disconnect_b() { - do_test_drop_messages_peer_disconnect(3); do_test_drop_messages_peer_disconnect(4); do_test_drop_messages_peer_disconnect(5); + do_test_drop_messages_peer_disconnect(6); } #[test] @@ -5353,9 +6185,8 @@ mod tests { let events_1 = nodes[0].node.get_and_clear_pending_msg_events(); assert_eq!(events_1.len(), 1); match events_1[0] { - MessageSendEvent::SendFundingLocked { ref node_id, msg: _, ref announcement_sigs } => { + MessageSendEvent::SendFundingLocked { ref node_id, msg: _ } => { assert_eq!(*node_id, nodes[1].node.get_our_node_id()); - assert!(announcement_sigs.is_none()); }, _ => panic!("Unexpected event"), } @@ -5364,9 +6195,8 @@ mod tests { let events_2 = nodes[1].node.get_and_clear_pending_msg_events(); assert_eq!(events_2.len(), 1); match events_2[0] { - MessageSendEvent::SendFundingLocked { ref node_id, msg: _, ref announcement_sigs } => { + MessageSendEvent::SendFundingLocked { ref node_id, msg: _ } => { assert_eq!(*node_id, nodes[0].node.get_our_node_id()); - assert!(announcement_sigs.is_none()); }, _ => panic!("Unexpected event"), } @@ -5431,8 +6261,9 @@ mod tests { _ => panic!("Unexpected event"), } - let (_, commitment_update) = nodes[0].node.handle_commitment_signed(&nodes[1].node.get_our_node_id(), commitment_signed).unwrap(); - assert!(commitment_update.is_none()); + nodes[0].node.handle_commitment_signed(&nodes[1].node.get_our_node_id(), commitment_signed).unwrap(); + let _ = get_event_msg!(nodes[0], MessageSendEvent::SendRevokeAndACK, nodes[1].node.get_our_node_id()); + // No commitment_signed so get_event_msg's assert(len == 1) passes check_added_monitors!(nodes[0], 1); }, _ => panic!("Unexpected event"), @@ -5441,13 +6272,17 @@ mod tests { 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); - let reestablish_1 = nodes[0].node.peer_connected(&nodes[1].node.get_our_node_id()); + nodes[0].node.peer_connected(&nodes[1].node.get_our_node_id()); + let reestablish_1 = get_chan_reestablish_msgs!(nodes[0], nodes[1]); assert_eq!(reestablish_1.len(), 1); - let reestablish_2 = nodes[1].node.peer_connected(&nodes[0].node.get_our_node_id()); + nodes[1].node.peer_connected(&nodes[0].node.get_our_node_id()); + let reestablish_2 = get_chan_reestablish_msgs!(nodes[1], nodes[0]); assert_eq!(reestablish_2.len(), 1); - let as_resp = nodes[0].node.handle_channel_reestablish(&nodes[1].node.get_our_node_id(), &reestablish_2[0]).unwrap(); - let bs_resp = nodes[1].node.handle_channel_reestablish(&nodes[0].node.get_our_node_id(), &reestablish_1[0]).unwrap(); + nodes[0].node.handle_channel_reestablish(&nodes[1].node.get_our_node_id(), &reestablish_2[0]).unwrap(); + let as_resp = handle_chan_reestablish_msgs!(nodes[0], nodes[1]); + nodes[1].node.handle_channel_reestablish(&nodes[0].node.get_our_node_id(), &reestablish_1[0]).unwrap(); + let bs_resp = handle_chan_reestablish_msgs!(nodes[1], nodes[0]); assert!(as_resp.0.is_none()); assert!(bs_resp.0.is_none()); @@ -5455,7 +6290,7 @@ mod tests { assert!(bs_resp.1.is_none()); assert!(bs_resp.2.is_none()); - assert!(as_resp.3 == msgs::RAACommitmentOrder::CommitmentFirst); + assert!(as_resp.3 == RAACommitmentOrder::CommitmentFirst); assert_eq!(as_resp.2.as_ref().unwrap().update_add_htlcs.len(), 1); assert!(as_resp.2.as_ref().unwrap().update_fulfill_htlcs.is_empty()); @@ -5463,11 +6298,13 @@ mod tests { assert!(as_resp.2.as_ref().unwrap().update_fail_malformed_htlcs.is_empty()); assert!(as_resp.2.as_ref().unwrap().update_fee.is_none()); nodes[1].node.handle_update_add_htlc(&nodes[0].node.get_our_node_id(), &as_resp.2.as_ref().unwrap().update_add_htlcs[0]).unwrap(); - let (bs_revoke_and_ack, bs_commitment_signed) = nodes[1].node.handle_commitment_signed(&nodes[0].node.get_our_node_id(), &as_resp.2.as_ref().unwrap().commitment_signed).unwrap(); - assert!(bs_commitment_signed.is_none()); + nodes[1].node.handle_commitment_signed(&nodes[0].node.get_our_node_id(), &as_resp.2.as_ref().unwrap().commitment_signed).unwrap(); + let bs_revoke_and_ack = get_event_msg!(nodes[1], MessageSendEvent::SendRevokeAndACK, nodes[0].node.get_our_node_id()); + // No commitment_signed so get_event_msg's assert(len == 1) passes check_added_monitors!(nodes[1], 1); - let bs_second_commitment_signed = nodes[1].node.handle_revoke_and_ack(&nodes[0].node.get_our_node_id(), as_resp.1.as_ref().unwrap()).unwrap().unwrap(); + nodes[1].node.handle_revoke_and_ack(&nodes[0].node.get_our_node_id(), as_resp.1.as_ref().unwrap()).unwrap(); + let bs_second_commitment_signed = get_htlc_update_msgs!(nodes[1], nodes[0].node.get_our_node_id()); assert!(bs_second_commitment_signed.update_add_htlcs.is_empty()); assert!(bs_second_commitment_signed.update_fulfill_htlcs.is_empty()); assert!(bs_second_commitment_signed.update_fail_htlcs.is_empty()); @@ -5475,7 +6312,8 @@ mod tests { assert!(bs_second_commitment_signed.update_fee.is_none()); check_added_monitors!(nodes[1], 1); - let as_commitment_signed = nodes[0].node.handle_revoke_and_ack(&nodes[1].node.get_our_node_id(), &bs_revoke_and_ack).unwrap().unwrap(); + nodes[0].node.handle_revoke_and_ack(&nodes[1].node.get_our_node_id(), &bs_revoke_and_ack).unwrap(); + let as_commitment_signed = get_htlc_update_msgs!(nodes[0], nodes[1].node.get_our_node_id()); assert!(as_commitment_signed.update_add_htlcs.is_empty()); assert!(as_commitment_signed.update_fulfill_htlcs.is_empty()); assert!(as_commitment_signed.update_fail_htlcs.is_empty()); @@ -5483,15 +6321,18 @@ mod tests { assert!(as_commitment_signed.update_fee.is_none()); check_added_monitors!(nodes[0], 1); - let (as_revoke_and_ack, as_second_commitment_signed) = nodes[0].node.handle_commitment_signed(&nodes[1].node.get_our_node_id(), &bs_second_commitment_signed.commitment_signed).unwrap(); - assert!(as_second_commitment_signed.is_none()); + nodes[0].node.handle_commitment_signed(&nodes[1].node.get_our_node_id(), &bs_second_commitment_signed.commitment_signed).unwrap(); + let as_revoke_and_ack = get_event_msg!(nodes[0], MessageSendEvent::SendRevokeAndACK, nodes[1].node.get_our_node_id()); + // No commitment_signed so get_event_msg's assert(len == 1) passes check_added_monitors!(nodes[0], 1); - let (bs_second_revoke_and_ack, bs_third_commitment_signed) = nodes[1].node.handle_commitment_signed(&nodes[0].node.get_our_node_id(), &as_commitment_signed.commitment_signed).unwrap(); - assert!(bs_third_commitment_signed.is_none()); + nodes[1].node.handle_commitment_signed(&nodes[0].node.get_our_node_id(), &as_commitment_signed.commitment_signed).unwrap(); + let bs_second_revoke_and_ack = get_event_msg!(nodes[1], MessageSendEvent::SendRevokeAndACK, nodes[0].node.get_our_node_id()); + // No commitment_signed so get_event_msg's assert(len == 1) passes check_added_monitors!(nodes[1], 1); - assert!(nodes[1].node.handle_revoke_and_ack(&nodes[0].node.get_our_node_id(), &as_revoke_and_ack).unwrap().is_none()); + nodes[1].node.handle_revoke_and_ack(&nodes[0].node.get_our_node_id(), &as_revoke_and_ack).unwrap(); + assert!(nodes[1].node.get_and_clear_pending_msg_events().is_empty()); check_added_monitors!(nodes[1], 1); let events_4 = nodes[1].node.get_and_clear_pending_events(); @@ -5513,7 +6354,8 @@ mod tests { _ => panic!("Unexpected event"), } - assert!(nodes[0].node.handle_revoke_and_ack(&nodes[1].node.get_our_node_id(), &bs_second_revoke_and_ack).unwrap().is_none()); + nodes[0].node.handle_revoke_and_ack(&nodes[1].node.get_our_node_id(), &bs_second_revoke_and_ack).unwrap(); + assert!(nodes[0].node.get_and_clear_pending_msg_events().is_empty()); check_added_monitors!(nodes[0], 1); claim_payment(&nodes[0], &[&nodes[1]], payment_preimage_2); @@ -5720,13 +6562,17 @@ mod tests { 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); - let reestablish_1 = nodes[0].node.peer_connected(&nodes[1].node.get_our_node_id()); + nodes[0].node.peer_connected(&nodes[1].node.get_our_node_id()); + let reestablish_1 = get_chan_reestablish_msgs!(nodes[0], nodes[1]); assert_eq!(reestablish_1.len(), 1); - let reestablish_2 = nodes[1].node.peer_connected(&nodes[0].node.get_our_node_id()); + nodes[1].node.peer_connected(&nodes[0].node.get_our_node_id()); + let reestablish_2 = get_chan_reestablish_msgs!(nodes[1], nodes[0]); assert_eq!(reestablish_2.len(), 1); - let as_resp = nodes[0].node.handle_channel_reestablish(&nodes[1].node.get_our_node_id(), &reestablish_2[0]).unwrap(); - let bs_resp = nodes[1].node.handle_channel_reestablish(&nodes[0].node.get_our_node_id(), &reestablish_1[0]).unwrap(); + nodes[0].node.handle_channel_reestablish(&nodes[1].node.get_our_node_id(), &reestablish_2[0]).unwrap(); + let as_resp = handle_chan_reestablish_msgs!(nodes[0], nodes[1]); + nodes[1].node.handle_channel_reestablish(&nodes[0].node.get_our_node_id(), &reestablish_1[0]).unwrap(); + let bs_resp = handle_chan_reestablish_msgs!(nodes[1], nodes[0]); assert!(as_resp.0.is_none()); assert!(bs_resp.0.is_none()); @@ -5738,15 +6584,19 @@ mod tests { assert!(nodes[0].node.get_and_clear_pending_events().is_empty()); assert!(nodes[0].node.get_and_clear_pending_msg_events().is_empty()); - let reestablish_1 = nodes[0].node.peer_connected(&nodes[1].node.get_our_node_id()); + nodes[0].node.peer_connected(&nodes[1].node.get_our_node_id()); + let reestablish_1 = get_chan_reestablish_msgs!(nodes[0], nodes[1]); assert_eq!(reestablish_1.len(), 1); - let reestablish_2 = nodes[1].node.peer_connected(&nodes[0].node.get_our_node_id()); + nodes[1].node.peer_connected(&nodes[0].node.get_our_node_id()); + let reestablish_2 = get_chan_reestablish_msgs!(nodes[1], nodes[0]); assert_eq!(reestablish_2.len(), 1); - let mut as_resp = nodes[0].node.handle_channel_reestablish(&nodes[1].node.get_our_node_id(), &reestablish_2[0]).unwrap(); + nodes[0].node.handle_channel_reestablish(&nodes[1].node.get_our_node_id(), &reestablish_2[0]).unwrap(); check_added_monitors!(nodes[0], 0); - let mut bs_resp = nodes[1].node.handle_channel_reestablish(&nodes[0].node.get_our_node_id(), &reestablish_1[0]).unwrap(); + let mut as_resp = handle_chan_reestablish_msgs!(nodes[0], nodes[1]); + nodes[1].node.handle_channel_reestablish(&nodes[0].node.get_our_node_id(), &reestablish_1[0]).unwrap(); check_added_monitors!(nodes[1], 0); + let mut bs_resp = handle_chan_reestablish_msgs!(nodes[1], nodes[0]); assert!(as_resp.0.is_none()); assert!(bs_resp.0.is_none()); @@ -5757,7 +6607,7 @@ mod tests { assert!(as_resp.1.is_some()); assert!(as_resp.2.is_some()); - assert!(as_resp.3 == msgs::RAACommitmentOrder::CommitmentFirst); + assert!(as_resp.3 == RAACommitmentOrder::CommitmentFirst); } else { assert!(bs_resp.2.as_ref().unwrap().update_add_htlcs.is_empty()); assert!(bs_resp.2.as_ref().unwrap().update_fail_htlcs.is_empty()); @@ -5778,8 +6628,9 @@ mod tests { _ => panic!("Unexpected event"), } - let (as_resp_raa, as_resp_cu) = nodes[0].node.handle_commitment_signed(&nodes[1].node.get_our_node_id(), &bs_resp.2.as_ref().unwrap().commitment_signed).unwrap(); - assert!(as_resp_cu.is_none()); + nodes[0].node.handle_commitment_signed(&nodes[1].node.get_our_node_id(), &bs_resp.2.as_ref().unwrap().commitment_signed).unwrap(); + let as_resp_raa = get_event_msg!(nodes[0], MessageSendEvent::SendRevokeAndACK, nodes[1].node.get_our_node_id()); + // No commitment_signed so get_event_msg's assert(len == 1) passes check_added_monitors!(nodes[0], 1); as_resp.1 = Some(as_resp_raa); @@ -5813,8 +6664,9 @@ mod tests { assert_eq!(payment_event.node_id, nodes[1].node.get_our_node_id()); nodes[1].node.handle_update_add_htlc(&nodes[0].node.get_our_node_id(), &payment_event.msgs[0]).unwrap(); - let (bs_revoke_and_ack, bs_commitment_signed) = nodes[1].node.handle_commitment_signed(&nodes[0].node.get_our_node_id(), &payment_event.commitment_msg).unwrap(); - assert!(bs_commitment_signed.is_none()); // nodes[1] is awaiting an RAA from nodes[0] still + nodes[1].node.handle_commitment_signed(&nodes[0].node.get_our_node_id(), &payment_event.commitment_msg).unwrap(); + let bs_revoke_and_ack = get_event_msg!(nodes[1], MessageSendEvent::SendRevokeAndACK, nodes[0].node.get_our_node_id()); + // nodes[1] is awaiting an RAA from nodes[0] still so get_event_msg's assert(len == 1) passes check_added_monitors!(nodes[1], 1); if disconnect_count & !disconnect_flags > 2 { @@ -5831,7 +6683,8 @@ mod tests { let bs_second_commitment_update; macro_rules! handle_bs_raa { () => { - as_commitment_update = nodes[0].node.handle_revoke_and_ack(&nodes[1].node.get_our_node_id(), &bs_revoke_and_ack).unwrap().unwrap(); + nodes[0].node.handle_revoke_and_ack(&nodes[1].node.get_our_node_id(), &bs_revoke_and_ack).unwrap(); + as_commitment_update = get_htlc_update_msgs!(nodes[0], nodes[1].node.get_our_node_id()); assert!(as_commitment_update.update_add_htlcs.is_empty()); assert!(as_commitment_update.update_fulfill_htlcs.is_empty()); assert!(as_commitment_update.update_fail_htlcs.is_empty()); @@ -5841,7 +6694,8 @@ mod tests { } } macro_rules! handle_initial_raa { () => { - bs_second_commitment_update = nodes[1].node.handle_revoke_and_ack(&nodes[0].node.get_our_node_id(), &initial_revoke_and_ack).unwrap().unwrap(); + nodes[1].node.handle_revoke_and_ack(&nodes[0].node.get_our_node_id(), &initial_revoke_and_ack).unwrap(); + bs_second_commitment_update = get_htlc_update_msgs!(nodes[1], nodes[0].node.get_our_node_id()); assert!(bs_second_commitment_update.update_add_htlcs.is_empty()); assert!(bs_second_commitment_update.update_fulfill_htlcs.is_empty()); assert!(bs_second_commitment_update.update_fail_htlcs.is_empty()); @@ -5862,7 +6716,7 @@ mod tests { assert!(as_resp.2.unwrap() == as_commitment_update); assert!(bs_resp.2.is_none()); - assert!(as_resp.3 == msgs::RAACommitmentOrder::RevokeAndACKFirst); + assert!(as_resp.3 == RAACommitmentOrder::RevokeAndACKFirst); } handle_initial_raa!(); @@ -5888,7 +6742,7 @@ mod tests { assert!(as_resp.2.is_none()); assert!(bs_resp.2.unwrap() == bs_second_commitment_update); - assert!(bs_resp.3 == msgs::RAACommitmentOrder::RevokeAndACKFirst); + assert!(bs_resp.3 == RAACommitmentOrder::RevokeAndACKFirst); } handle_bs_raa!(); @@ -5904,18 +6758,22 @@ mod tests { } } - let (as_revoke_and_ack, as_commitment_signed) = nodes[0].node.handle_commitment_signed(&nodes[1].node.get_our_node_id(), &bs_second_commitment_update.commitment_signed).unwrap(); - assert!(as_commitment_signed.is_none()); + nodes[0].node.handle_commitment_signed(&nodes[1].node.get_our_node_id(), &bs_second_commitment_update.commitment_signed).unwrap(); + let as_revoke_and_ack = get_event_msg!(nodes[0], MessageSendEvent::SendRevokeAndACK, nodes[1].node.get_our_node_id()); + // No commitment_signed so get_event_msg's assert(len == 1) passes check_added_monitors!(nodes[0], 1); - let (bs_second_revoke_and_ack, bs_third_commitment_signed) = nodes[1].node.handle_commitment_signed(&nodes[0].node.get_our_node_id(), &as_commitment_update.commitment_signed).unwrap(); - assert!(bs_third_commitment_signed.is_none()); + nodes[1].node.handle_commitment_signed(&nodes[0].node.get_our_node_id(), &as_commitment_update.commitment_signed).unwrap(); + let bs_second_revoke_and_ack = get_event_msg!(nodes[1], MessageSendEvent::SendRevokeAndACK, nodes[0].node.get_our_node_id()); + // No commitment_signed so get_event_msg's assert(len == 1) passes check_added_monitors!(nodes[1], 1); - assert!(nodes[1].node.handle_revoke_and_ack(&nodes[0].node.get_our_node_id(), &as_revoke_and_ack).unwrap().is_none()); + nodes[1].node.handle_revoke_and_ack(&nodes[0].node.get_our_node_id(), &as_revoke_and_ack).unwrap(); + assert!(nodes[1].node.get_and_clear_pending_msg_events().is_empty()); check_added_monitors!(nodes[1], 1); - assert!(nodes[0].node.handle_revoke_and_ack(&nodes[1].node.get_our_node_id(), &bs_second_revoke_and_ack).unwrap().is_none()); + nodes[0].node.handle_revoke_and_ack(&nodes[1].node.get_our_node_id(), &bs_second_revoke_and_ack).unwrap(); + assert!(nodes[0].node.get_and_clear_pending_msg_events().is_empty()); check_added_monitors!(nodes[0], 1); expect_pending_htlcs_forwardable!(nodes[1]); @@ -6033,4 +6891,203 @@ mod tests { sign_msg!(unsigned_msg); assert!(nodes[0].router.handle_channel_announcement(&chan_announcement).is_err()); } + + struct VecWriter(Vec); + impl Writer for VecWriter { + fn write_all(&mut self, buf: &[u8]) -> Result<(), ::std::io::Error> { + self.0.extend_from_slice(buf); + Ok(()) + } + fn size_hint(&mut self, size: usize) { + self.0.reserve_exact(size); + } + } + + #[test] + fn test_no_txn_manager_serialize_deserialize() { + let mut nodes = create_network(2); + + let tx = create_chan_between_nodes_with_value_init(&nodes[0], &nodes[1], 100000, 10001); + + nodes[1].node.peer_disconnected(&nodes[0].node.get_our_node_id(), false); + + let nodes_0_serialized = nodes[0].node.encode(); + let mut chan_0_monitor_serialized = VecWriter(Vec::new()); + nodes[0].chan_monitor.simple_monitor.monitors.lock().unwrap().iter().next().unwrap().1.write_for_disk(&mut chan_0_monitor_serialized).unwrap(); + + nodes[0].chan_monitor = Arc::new(test_utils::TestChannelMonitor::new(nodes[0].chain_monitor.clone(), nodes[0].tx_broadcaster.clone(), Arc::new(test_utils::TestLogger::new()))); + let mut chan_0_monitor_read = &chan_0_monitor_serialized.0[..]; + let (_, chan_0_monitor) = <(Sha256dHash, ChannelMonitor)>::read(&mut chan_0_monitor_read, Arc::new(test_utils::TestLogger::new())).unwrap(); + assert!(chan_0_monitor_read.is_empty()); + + let mut nodes_0_read = &nodes_0_serialized[..]; + let config = UserConfig::new(); + let keys_manager = Arc::new(keysinterface::KeysManager::new(&nodes[0].node_seed, Network::Testnet, Arc::new(test_utils::TestLogger::new()))); + let (_, nodes_0_deserialized) = { + let mut channel_monitors = HashMap::new(); + channel_monitors.insert(chan_0_monitor.get_funding_txo().unwrap(), &chan_0_monitor); + <(Sha256dHash, ChannelManager)>::read(&mut nodes_0_read, ChannelManagerReadArgs { + default_config: config, + keys_manager, + fee_estimator: Arc::new(test_utils::TestFeeEstimator { sat_per_kw: 253 }), + monitor: nodes[0].chan_monitor.clone(), + chain_monitor: nodes[0].chain_monitor.clone(), + tx_broadcaster: nodes[0].tx_broadcaster.clone(), + logger: Arc::new(test_utils::TestLogger::new()), + channel_monitors: &channel_monitors, + }).unwrap() + }; + assert!(nodes_0_read.is_empty()); + + assert!(nodes[0].chan_monitor.add_update_monitor(chan_0_monitor.get_funding_txo().unwrap(), chan_0_monitor).is_ok()); + nodes[0].node = Arc::new(nodes_0_deserialized); + let nodes_0_as_listener: Arc = nodes[0].node.clone(); + nodes[0].chain_monitor.register_listener(Arc::downgrade(&nodes_0_as_listener)); + assert_eq!(nodes[0].node.list_channels().len(), 1); + check_added_monitors!(nodes[0], 1); + + nodes[0].node.peer_connected(&nodes[1].node.get_our_node_id()); + let reestablish_1 = get_chan_reestablish_msgs!(nodes[0], nodes[1]); + nodes[1].node.peer_connected(&nodes[0].node.get_our_node_id()); + let reestablish_2 = get_chan_reestablish_msgs!(nodes[1], nodes[0]); + + nodes[1].node.handle_channel_reestablish(&nodes[0].node.get_our_node_id(), &reestablish_1[0]).unwrap(); + assert!(nodes[1].node.get_and_clear_pending_msg_events().is_empty()); + nodes[0].node.handle_channel_reestablish(&nodes[1].node.get_our_node_id(), &reestablish_2[0]).unwrap(); + assert!(nodes[0].node.get_and_clear_pending_msg_events().is_empty()); + + let (funding_locked, _) = create_chan_between_nodes_with_value_confirm(&nodes[0], &nodes[1], &tx); + let (announcement, as_update, bs_update) = create_chan_between_nodes_with_value_b(&nodes[0], &nodes[1], &funding_locked); + for node in nodes.iter() { + assert!(node.router.handle_channel_announcement(&announcement).unwrap()); + node.router.handle_channel_update(&as_update).unwrap(); + node.router.handle_channel_update(&bs_update).unwrap(); + } + + send_payment(&nodes[0], &[&nodes[1]], 1000000); + } + + #[test] + fn test_simple_manager_serialize_deserialize() { + let mut nodes = create_network(2); + create_announced_chan_between_nodes(&nodes, 0, 1); + + let (our_payment_preimage, _) = route_payment(&nodes[0], &[&nodes[1]], 1000000); + let (_, our_payment_hash) = route_payment(&nodes[0], &[&nodes[1]], 1000000); + + nodes[1].node.peer_disconnected(&nodes[0].node.get_our_node_id(), false); + + let nodes_0_serialized = nodes[0].node.encode(); + let mut chan_0_monitor_serialized = VecWriter(Vec::new()); + nodes[0].chan_monitor.simple_monitor.monitors.lock().unwrap().iter().next().unwrap().1.write_for_disk(&mut chan_0_monitor_serialized).unwrap(); + + nodes[0].chan_monitor = Arc::new(test_utils::TestChannelMonitor::new(nodes[0].chain_monitor.clone(), nodes[0].tx_broadcaster.clone(), Arc::new(test_utils::TestLogger::new()))); + let mut chan_0_monitor_read = &chan_0_monitor_serialized.0[..]; + let (_, chan_0_monitor) = <(Sha256dHash, ChannelMonitor)>::read(&mut chan_0_monitor_read, Arc::new(test_utils::TestLogger::new())).unwrap(); + assert!(chan_0_monitor_read.is_empty()); + + let mut nodes_0_read = &nodes_0_serialized[..]; + let keys_manager = Arc::new(keysinterface::KeysManager::new(&nodes[0].node_seed, Network::Testnet, Arc::new(test_utils::TestLogger::new()))); + let (_, nodes_0_deserialized) = { + let mut channel_monitors = HashMap::new(); + channel_monitors.insert(chan_0_monitor.get_funding_txo().unwrap(), &chan_0_monitor); + <(Sha256dHash, ChannelManager)>::read(&mut nodes_0_read, ChannelManagerReadArgs { + default_config: UserConfig::new(), + keys_manager, + fee_estimator: Arc::new(test_utils::TestFeeEstimator { sat_per_kw: 253 }), + monitor: nodes[0].chan_monitor.clone(), + chain_monitor: nodes[0].chain_monitor.clone(), + tx_broadcaster: nodes[0].tx_broadcaster.clone(), + logger: Arc::new(test_utils::TestLogger::new()), + channel_monitors: &channel_monitors, + }).unwrap() + }; + assert!(nodes_0_read.is_empty()); + + assert!(nodes[0].chan_monitor.add_update_monitor(chan_0_monitor.get_funding_txo().unwrap(), chan_0_monitor).is_ok()); + nodes[0].node = Arc::new(nodes_0_deserialized); + check_added_monitors!(nodes[0], 1); + + reconnect_nodes(&nodes[0], &nodes[1], false, (0, 0), (0, 0), (0, 0), (0, 0), (false, false)); + + fail_payment(&nodes[0], &[&nodes[1]], our_payment_hash); + claim_payment(&nodes[0], &[&nodes[1]], our_payment_preimage); + } + + #[test] + fn test_manager_serialize_deserialize_inconsistent_monitor() { + // Test deserializing a ChannelManager with a out-of-date ChannelMonitor + let mut nodes = create_network(4); + create_announced_chan_between_nodes(&nodes, 0, 1); + create_announced_chan_between_nodes(&nodes, 2, 0); + let (_, _, channel_id, funding_tx) = create_announced_chan_between_nodes(&nodes, 0, 3); + + let (our_payment_preimage, _) = route_payment(&nodes[2], &[&nodes[0], &nodes[1]], 1000000); + + // Serialize the ChannelManager here, but the monitor we keep up-to-date + let nodes_0_serialized = nodes[0].node.encode(); + + route_payment(&nodes[0], &[&nodes[3]], 1000000); + nodes[1].node.peer_disconnected(&nodes[0].node.get_our_node_id(), false); + nodes[2].node.peer_disconnected(&nodes[0].node.get_our_node_id(), false); + nodes[3].node.peer_disconnected(&nodes[0].node.get_our_node_id(), false); + + // Now the ChannelMonitor (which is now out-of-sync with ChannelManager for channel w/ + // nodes[3]) + let mut node_0_monitors_serialized = Vec::new(); + for monitor in nodes[0].chan_monitor.simple_monitor.monitors.lock().unwrap().iter() { + let mut writer = VecWriter(Vec::new()); + monitor.1.write_for_disk(&mut writer).unwrap(); + node_0_monitors_serialized.push(writer.0); + } + + nodes[0].chan_monitor = Arc::new(test_utils::TestChannelMonitor::new(nodes[0].chain_monitor.clone(), nodes[0].tx_broadcaster.clone(), Arc::new(test_utils::TestLogger::new()))); + let mut node_0_monitors = Vec::new(); + for serialized in node_0_monitors_serialized.iter() { + let mut read = &serialized[..]; + let (_, monitor) = <(Sha256dHash, ChannelMonitor)>::read(&mut read, Arc::new(test_utils::TestLogger::new())).unwrap(); + assert!(read.is_empty()); + node_0_monitors.push(monitor); + } + + let mut nodes_0_read = &nodes_0_serialized[..]; + let keys_manager = Arc::new(keysinterface::KeysManager::new(&nodes[0].node_seed, Network::Testnet, Arc::new(test_utils::TestLogger::new()))); + let (_, nodes_0_deserialized) = <(Sha256dHash, ChannelManager)>::read(&mut nodes_0_read, ChannelManagerReadArgs { + default_config: UserConfig::new(), + keys_manager, + fee_estimator: Arc::new(test_utils::TestFeeEstimator { sat_per_kw: 253 }), + monitor: nodes[0].chan_monitor.clone(), + chain_monitor: nodes[0].chain_monitor.clone(), + tx_broadcaster: nodes[0].tx_broadcaster.clone(), + logger: Arc::new(test_utils::TestLogger::new()), + channel_monitors: &node_0_monitors.iter().map(|monitor| { (monitor.get_funding_txo().unwrap(), monitor) }).collect(), + }).unwrap(); + assert!(nodes_0_read.is_empty()); + + { // Channel close should result in a commitment tx and an HTLC tx + let txn = nodes[0].tx_broadcaster.txn_broadcasted.lock().unwrap(); + assert_eq!(txn.len(), 2); + assert_eq!(txn[0].input[0].previous_output.txid, funding_tx.txid()); + assert_eq!(txn[1].input[0].previous_output.txid, txn[0].txid()); + } + + for monitor in node_0_monitors.drain(..) { + assert!(nodes[0].chan_monitor.add_update_monitor(monitor.get_funding_txo().unwrap(), monitor).is_ok()); + check_added_monitors!(nodes[0], 1); + } + nodes[0].node = Arc::new(nodes_0_deserialized); + + // nodes[1] and nodes[2] have no lost state with nodes[0]... + reconnect_nodes(&nodes[0], &nodes[1], false, (0, 0), (0, 0), (0, 0), (0, 0), (false, false)); + reconnect_nodes(&nodes[0], &nodes[2], false, (0, 0), (0, 0), (0, 0), (0, 0), (false, false)); + //... and we can even still claim the payment! + claim_payment(&nodes[2], &[&nodes[0], &nodes[1]], our_payment_preimage); + + nodes[3].node.peer_connected(&nodes[0].node.get_our_node_id()); + let reestablish = get_event_msg!(nodes[3], MessageSendEvent::SendChannelReestablish, nodes[0].node.get_our_node_id()); + nodes[0].node.peer_connected(&nodes[3].node.get_our_node_id()); + if let Err(msgs::HandleError { action: Some(msgs::ErrorAction::SendErrorMessage { msg }), .. }) = nodes[0].node.handle_channel_reestablish(&nodes[3].node.get_our_node_id(), &reestablish) { + assert_eq!(msg.channel_id, channel_id); + } else { panic!("Unexpected result"); } + } }