X-Git-Url: http://git.bitcoin.ninja/index.cgi?a=blobdiff_plain;f=src%2Fln%2Fchannelmanager.rs;h=83ff021a7e9164b121f1503b599cd62d07d852c8;hb=fd141bbed2a900070cbecf7ef9f22e377e1f353f;hp=f612955bec4fcd2f6ccd2b201bb861458892f89c;hpb=2d8afeccdb4f5aa6232e6d87c6096e980b836072;p=rust-lightning diff --git a/src/ln/channelmanager.rs b/src/ln/channelmanager.rs index f612955b..83ff021a 100644 --- a/src/ln/channelmanager.rs +++ b/src/ln/channelmanager.rs @@ -12,6 +12,7 @@ use secp256k1::ecdh::SharedSecret; use secp256k1; use chain::chaininterface::{BroadcasterInterface,ChainListener,ChainWatchInterface,FeeEstimator}; +use chain::transaction::OutPoint; use ln::channel::{Channel, ChannelKeys}; use ln::channelmonitor::ManyChannelMonitor; use ln::router::{Route,RouteHop}; @@ -166,7 +167,7 @@ macro_rules! secp_call { match $res { Ok(key) => key, //TODO: Make the err a parameter! - Err(_) => return Err(HandleError{err: "Key error", msg: None}) + Err(_) => return Err(HandleError{err: "Key error", action: None}) } }; } @@ -201,7 +202,7 @@ impl ChannelManager { /// 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 >= (1 << 24)! + /// panics if channel_value_satoshis is >= `MAX_FUNDING_SATOSHIS`! pub fn new(our_network_key: SecretKey, fee_proportional_millionths: u32, announce_channels_publicly: bool, network: Network, feeest: Arc, monitor: Arc, chain_monitor: Arc, tx_broadcaster: Arc) -> Result, secp256k1::Error> { let secp_ctx = Secp256k1::new(); @@ -232,7 +233,14 @@ impl ChannelManager { Ok(res) } - pub fn create_channel(&self, their_network_key: PublicKey, channel_value_satoshis: u64, user_id: u64) -> Result { + /// Creates a new outbound channel to the given remote node and with the given value. + /// user_id will be provided back as user_channel_id in FundingGenerationReady and + /// FundingBroadcastSafe events to allow tracking of which events correspond with which + /// create_channel call. Note that user_channel_id defaults to 0 for inbound channels, so you + /// may wish to avoid using 0 for user_id here. + /// If successful, will generate a SendOpenChannel event, so you should probably poll + /// PeerManager::process_events afterwards. + pub fn create_channel(&self, their_network_key: PublicKey, channel_value_satoshis: u64, user_id: u64) -> Result<(), HandleError> { let chan_keys = if cfg!(feature = "fuzztarget") { ChannelKeys { funding_key: SecretKey::from_slice(&self.secp_ctx, &[0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0]).unwrap(), @@ -258,8 +266,15 @@ impl ChannelManager { let mut channel_state = self.channel_state.lock().unwrap(); match channel_state.by_id.insert(channel.channel_id(), channel) { Some(_) => panic!("RNG is bad???"), - None => Ok(res) + None => {} } + + let mut events = self.pending_events.lock().unwrap(); + events.push(events::Event::SendOpenChannel { + node_id: their_network_key, + msg: res, + }); + Ok(()) } /// Gets the list of open channels, in random order. See ChannelDetail field documentation for @@ -283,10 +298,9 @@ impl ChannelManager { /// will be accepted on the given channel, and after additional timeout/the closing of all /// pending HTLCs, the channel will be closed on chain. pub fn close_channel(&self, channel_id: &Uint256) -> Result { - let res = { + let (res, 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(channel_id.clone()) { hash_map::Entry::Occupied(mut chan_entry) => { let res = chan_entry.get_mut().get_shutdown()?; @@ -294,17 +308,24 @@ impl ChannelManager { if let Some(short_id) = chan_entry.get().get_short_channel_id() { channel_state.short_to_id.remove(&short_id); } - chan_entry.remove_entry(); - } - res + (res, Some(chan_entry.remove_entry().1)) + } else { (res, None) } }, - hash_map::Entry::Vacant(_) => return Err(HandleError{err: "No such channel", msg: None}) + hash_map::Entry::Vacant(_) => return Err(HandleError{err: "No such channel", action: None}) } }; for payment_hash in res.1 { // unknown_next_peer...I dunno who that is anymore.... self.fail_htlc_backwards_internal(self.channel_state.lock().unwrap(), &payment_hash, HTLCFailReason::Reason { failure_code: 0x4000 | 10, data: Vec::new() }); } + if let Some(chan) = chan_option { + if let Ok(update) = self.get_channel_update(&chan) { + let mut events = self.pending_events.lock().unwrap(); + events.push(events::Event::BroadcastChannelUpdate { + msg: update + }); + } + } Ok(res.0) } @@ -422,11 +443,11 @@ impl ChannelManager { }; cur_value_msat += hop.fee_msat; if cur_value_msat >= 21000000 * 100000000 * 1000 { - return Err(HandleError{err: "Channel fees overflowed?!", msg: None}); + return Err(HandleError{err: "Channel fees overflowed?!", action: None}); } cur_cltv += hop.cltv_expiry_delta as u32; if cur_cltv >= 500000000 { - return Err(HandleError{err: "Channel CLTV overflowed?!", msg: None}); + return Err(HandleError{err: "Channel CLTV overflowed?!", action: None}); } last_short_channel_id = hop.short_channel_id; } @@ -553,9 +574,9 @@ impl ChannelManager { } /// only fails if the channel does not yet have an assigned short_id - fn get_channel_update(&self, chan: &mut Channel) -> Result { + fn get_channel_update(&self, chan: &Channel) -> Result { let short_channel_id = match chan.get_short_channel_id() { - None => return Err(HandleError{err: "Channel not yet established", msg: None}), + None => return Err(HandleError{err: "Channel not yet established", action: None}), Some(id) => id, }; @@ -581,18 +602,19 @@ impl ChannelManager { }) } - /// Sends a payment along a given route, returning the UpdateAddHTLC message to give to the - /// first hop in route. Value parameters are provided via the last hop in route, see - /// documentation for RouteHop fields for more info. + /// Sends a payment along a given route. + /// Value parameters are provided via the last hop in route, see documentation for RouteHop + /// fields for more info. /// See-also docs on Channel::send_htlc_and_commit. - pub fn send_payment(&self, route: Route, payment_hash: [u8; 32]) -> Result, HandleError> { + /// May generate a SendHTLCs event on success, which should be relayed. + pub fn send_payment(&self, route: Route, payment_hash: [u8; 32]) -> Result<(), HandleError> { if route.hops.len() < 1 || route.hops.len() > 20 { - return Err(HandleError{err: "Route didn't go anywhere/had bogus size", msg: None}); + return Err(HandleError{err: "Route didn't go anywhere/had bogus size", action: None}); } let our_node_id = self.get_our_node_id(); for (idx, hop) in route.hops.iter().enumerate() { if idx != route.hops.len() - 1 && hop.pubkey == our_node_id { - return Err(HandleError{err: "Route went through us but wasn't a simple rebalance loop to us", msg: None}); + return Err(HandleError{err: "Route went through us but wasn't a simple rebalance loop to us", action: None}); } } @@ -608,20 +630,22 @@ impl ChannelManager { let (onion_payloads, htlc_msat, htlc_cltv) = ChannelManager::build_onion_payloads(&route)?; let onion_packet = ChannelManager::construct_onion_packet(onion_payloads, onion_keys, associated_data)?; - let (update_add, commitment_signed, chan_monitor) = { + let (first_hop_node_id, (update_add, commitment_signed, chan_monitor)) = { 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) { - None => return Err(HandleError{err: "No channel available with first hop!", msg: None}), + None => return Err(HandleError{err: "No channel available with first hop!", action: None}), Some(id) => id.clone() }; let res = { let chan = channel_state.by_id.get_mut(&id).unwrap(); if chan.get_their_node_id() != route.hops.first().unwrap().pubkey { - return Err(HandleError{err: "Node ID mismatch on first hop!", msg: None}); + return Err(HandleError{err: "Node ID mismatch on first hop!", action: None}); } chan.send_htlc_and_commit(htlc_msat, payment_hash.clone(), htlc_cltv, onion_packet)? }; + let first_hop_node_id = route.hops.first().unwrap().pubkey; + if channel_state.claimable_htlcs.insert(payment_hash, PendingOutboundHTLC::OutboundRoute { route, session_priv, @@ -632,25 +656,32 @@ impl ChannelManager { } match res { - Some(msgs) => msgs, - None => return Ok(None), + Some(msgs) => (first_hop_node_id, msgs), + None => return Ok(()), } }; if let Err(_e) = self.monitor.add_update_monitor(chan_monitor.get_funding_txo().unwrap(), chan_monitor) { unimplemented!(); // maybe remove from claimable_htlcs? } - Ok(Some((update_add, commitment_signed))) + + let mut events = self.pending_events.lock().unwrap(); + events.push(events::Event::SendHTLCs { + node_id: first_hop_node_id, + msgs: vec![update_add], + commitment_msg: commitment_signed, + }); + Ok(()) } /// Call this upon creation of a funding transaction for the given channel. /// Panics if a funding transaction has already been provided for this channel. - pub fn funding_transaction_generated(&self, temporary_channel_id: &Uint256, funding_txo: (Sha256dHash, u16)) { + pub fn funding_transaction_generated(&self, temporary_channel_id: &Uint256, funding_txo: OutPoint) { let (chan, msg, chan_monitor) = { let mut channel_state = self.channel_state.lock().unwrap(); match channel_state.by_id.remove(&temporary_channel_id) { Some(mut chan) => { - match chan.get_outbound_funding_created(funding_txo.0, funding_txo.1) { + match chan.get_outbound_funding_created(funding_txo) { Ok(funding_msg) => { (chan, funding_msg.0, funding_msg.1) }, @@ -713,7 +744,6 @@ impl ChannelManager { for forward_info in pending_forwards { failed_forwards.push((forward_info.payment_hash, 0x4000 | 10, None)); } - // TODO: Send a failure packet back on each pending_forward continue; } }; @@ -724,7 +754,7 @@ impl ChannelManager { match forward_chan.send_htlc(forward_info.amt_to_forward, forward_info.payment_hash, forward_info.outgoing_cltv_value, forward_info.onion_packet.unwrap()) { Err(_e) => { let chan_update = self.get_channel_update(forward_chan).unwrap(); - failed_forwards.push((forward_info.payment_hash, 0x4000 | 7, Some(chan_update))); + failed_forwards.push((forward_info.payment_hash, 0x1000 | 7, Some(chan_update))); continue; }, Ok(update_add) => { @@ -989,7 +1019,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 mut new_funding_locked_messages = Vec::new(); + let mut new_events = Vec::new(); { let mut channel_state = self.channel_state.lock().unwrap(); let mut short_to_ids_to_insert = Vec::new(); @@ -1003,7 +1033,7 @@ impl ChainListener for ChannelManager { return true; } }; - new_funding_locked_messages.push(events::Event::SendFundingLocked { + new_events.push(events::Event::SendFundingLocked { node_id: channel.get_their_node_id(), msg: funding_locked, announcement_sigs: announcement_sigs @@ -1013,11 +1043,16 @@ impl ChainListener for ChannelManager { if let Some(funding_txo) = channel.get_funding_txo() { for tx in txn_matched { for inp in tx.input.iter() { - if inp.prev_hash == funding_txo.0 && inp.prev_index == funding_txo.1 as u32 { + if inp.prev_hash == funding_txo.txid && inp.prev_index == funding_txo.index as u32 { if let Some(short_id) = channel.get_short_channel_id() { short_to_ids_to_remove.push(short_id); } channel.force_shutdown(); + if let Ok(update) = self.get_channel_update(&channel) { + new_events.push(events::Event::BroadcastChannelUpdate { + msg: update + }); + } return false; } } @@ -1028,6 +1063,11 @@ impl ChainListener for ChannelManager { short_to_ids_to_remove.push(short_id); } channel.force_shutdown(); + if let Ok(update) = self.get_channel_update(&channel) { + new_events.push(events::Event::BroadcastChannelUpdate { + msg: update + }); + } return false; } true @@ -1040,7 +1080,7 @@ impl ChainListener for ChannelManager { } } let mut pending_events = self.pending_events.lock().unwrap(); - for funding_locked in new_funding_locked_messages.drain(..) { + for funding_locked in new_events.drain(..) { pending_events.push(funding_locked); } } @@ -1059,11 +1099,11 @@ 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 { if msg.chain_hash != self.genesis_hash { - return Err(HandleError{err: "Unknown genesis block hash", msg: None}); + return Err(HandleError{err: "Unknown genesis block hash", action: None}); } let mut channel_state = self.channel_state.lock().unwrap(); if channel_state.by_id.contains_key(&msg.temporary_channel_id) { - return Err(HandleError{err: "temporary_channel_id collision!", msg: None}); + return Err(HandleError{err: "temporary_channel_id collision!", action: None}); } let chan_keys = if cfg!(feature = "fuzztarget") { @@ -1098,12 +1138,12 @@ impl ChannelMessageHandler for ChannelManager { match channel_state.by_id.get_mut(&msg.temporary_channel_id) { Some(chan) => { if chan.get_their_node_id() != *their_node_id { - return Err(HandleError{err: "Got a message for a channel from the wrong node!", msg: None}) + return Err(HandleError{err: "Got a message for a channel from the wrong node!", action: None}) } chan.accept_channel(&msg)?; (chan.get_value_satoshis(), chan.get_funding_redeemscript().to_v0_p2wsh(), chan.get_user_id()) }, - None => return Err(HandleError{err: "Failed to find corresponding channel", msg: None}) + None => return Err(HandleError{err: "Failed to find corresponding channel", action: None}) } }; let mut pending_events = self.pending_events.lock().unwrap(); @@ -1125,7 +1165,7 @@ impl ChannelMessageHandler for ChannelManager { match channel_state.by_id.remove(&msg.temporary_channel_id) { Some(mut chan) => { if chan.get_their_node_id() != *their_node_id { - return Err(HandleError{err: "Got a message for a channel from the wrong node!", msg: None}) + return Err(HandleError{err: "Got a message for a channel from the wrong node!", action: None}) } match chan.funding_created(msg) { Ok((funding_msg, monitor_update)) => { @@ -1136,7 +1176,7 @@ impl ChannelMessageHandler for ChannelManager { } } }, - None => return Err(HandleError{err: "Failed to find corresponding channel", msg: None}) + None => return Err(HandleError{err: "Failed to find corresponding channel", action: None}) } }; // Release channel lock for install_watch_outpoint call, // note that this means if the remote end is misbehaving and sends a message for the same @@ -1156,12 +1196,12 @@ impl ChannelMessageHandler for ChannelManager { match channel_state.by_id.get_mut(&msg.channel_id) { Some(chan) => { if chan.get_their_node_id() != *their_node_id { - return Err(HandleError{err: "Got a message for a channel from the wrong node!", msg: None}) + return Err(HandleError{err: "Got a message for a channel from the wrong node!", action: None}) } let chan_monitor = chan.funding_signed(&msg)?; (chan.get_funding_txo().unwrap(), chan.get_user_id(), chan_monitor) }, - None => return Err(HandleError{err: "Failed to find corresponding channel", msg: None}) + None => return Err(HandleError{err: "Failed to find corresponding channel", action: None}) } }; if let Err(_e) = self.monitor.add_update_monitor(monitor.get_funding_txo().unwrap(), monitor) { @@ -1180,52 +1220,59 @@ impl ChannelMessageHandler for ChannelManager { match channel_state.by_id.get_mut(&msg.channel_id) { Some(chan) => { if chan.get_their_node_id() != *their_node_id { - return Err(HandleError{err: "Got a message for a channel from the wrong node!", msg: None}) + return Err(HandleError{err: "Got a message for a channel from the wrong node!", action: None}) } chan.funding_locked(&msg)?; return Ok(self.get_announcement_sigs(chan)?); }, - None => return Err(HandleError{err: "Failed to find corresponding channel", msg: None}) + None => return Err(HandleError{err: "Failed to find corresponding channel", action: None}) }; } fn handle_shutdown(&self, their_node_id: &PublicKey, msg: &msgs::Shutdown) -> Result<(Option, Option), HandleError> { - let res = { + let (res, 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()) { hash_map::Entry::Occupied(mut chan_entry) => { if chan_entry.get().get_their_node_id() != *their_node_id { - return Err(HandleError{err: "Got a message for a channel from the wrong node!", msg: None}) + return Err(HandleError{err: "Got a message for a channel from the wrong node!", action: None}) } let res = chan_entry.get_mut().shutdown(&*self.fee_estimator, &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); } - chan_entry.remove_entry(); - } - res + (res, Some(chan_entry.remove_entry().1)) + } else { (res, None) } }, - hash_map::Entry::Vacant(_) => return Err(HandleError{err: "Failed to find corresponding channel", msg: None}) + hash_map::Entry::Vacant(_) => return Err(HandleError{err: "Failed to find corresponding channel", action: None}) } }; for payment_hash in res.2 { // unknown_next_peer...I dunno who that is anymore.... self.fail_htlc_backwards_internal(self.channel_state.lock().unwrap(), &payment_hash, HTLCFailReason::Reason { failure_code: 0x4000 | 10, data: Vec::new() }); } + if let Some(chan) = chan_option { + if let Ok(update) = self.get_channel_update(&chan) { + let mut events = self.pending_events.lock().unwrap(); + events.push(events::Event::BroadcastChannelUpdate { + msg: update + }); + } + } Ok((res.0, res.1)) } fn handle_closing_signed(&self, their_node_id: &PublicKey, msg: &msgs::ClosingSigned) -> Result, HandleError> { - let res = { + let (res, 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()) { hash_map::Entry::Occupied(mut chan_entry) => { if chan_entry.get().get_their_node_id() != *their_node_id { - return Err(HandleError{err: "Got a message for a channel from the wrong node!", msg: None}) + return Err(HandleError{err: "Got a message for a channel from the wrong node!", action: None}) } let res = chan_entry.get_mut().closing_signed(&*self.fee_estimator, &msg)?; if res.1.is_some() { @@ -1237,16 +1284,23 @@ impl ChannelMessageHandler for ChannelManager { if let Some(short_id) = chan_entry.get().get_short_channel_id() { channel_state.short_to_id.remove(&short_id); } - chan_entry.remove_entry(); - } - res + (res, Some(chan_entry.remove_entry().1)) + } else { (res, None) } }, - hash_map::Entry::Vacant(_) => return Err(HandleError{err: "Failed to find corresponding channel", msg: None}) + hash_map::Entry::Vacant(_) => return Err(HandleError{err: "Failed to find corresponding channel", action: None}) } }; if let Some(broadcast_tx) = res.1 { self.tx_broadcaster.broadcast_transaction(&broadcast_tx); } + if let Some(chan) = chan_option { + if let Ok(update) = self.get_channel_update(&chan) { + let mut events = self.pending_events.lock().unwrap(); + events.push(events::Event::BroadcastChannelUpdate { + msg: update + }); + } + } Ok(res.0) } @@ -1281,7 +1335,7 @@ impl ChannelMessageHandler for ChannelManager { ($msg: expr, $err_code: expr, $data: expr) => { return Err(msgs::HandleError { err: $msg, - msg: Some(msgs::ErrorAction::UpdateFailHTLC { + action: Some(msgs::ErrorAction::UpdateFailHTLC { msg: msgs::UpdateFailHTLC { channel_id: msg.channel_id, htlc_id: msg.htlc_id, @@ -1411,7 +1465,7 @@ impl ChannelMessageHandler for ChannelManager { let chan = channel_state.by_id.get_mut(&forwarding_id).unwrap(); if !chan.is_live() { let chan_update = self.get_channel_update(chan).unwrap(); - return_err!("Forwarding channel is not in a ready state.", 0x4000 | 7, &chan_update.encode_with_len()[..]); + return_err!("Forwarding channel is not in a ready state.", 0x1000 | 7, &chan_update.encode_with_len()[..]); } } @@ -1440,16 +1494,16 @@ impl ChannelMessageHandler for ChannelManager { let (source_short_channel_id, res) = match channel_state.by_id.get_mut(&msg.channel_id) { Some(chan) => { if chan.get_their_node_id() != *their_node_id { - return Err(HandleError{err: "Got a message for a channel from the wrong node!", msg: None}) + return Err(HandleError{err: "Got a message for a channel from the wrong node!", action: None}) } if !chan.is_usable() { - return Err(HandleError{err: "Channel not yet available for receiving HTLCs", msg: None}); + return Err(HandleError{err: "Channel not yet available for receiving HTLCs", action: None}); } let short_channel_id = chan.get_short_channel_id().unwrap(); pending_forward_info.prev_short_channel_id = short_channel_id; (short_channel_id, chan.update_add_htlc(&msg, pending_forward_info)?) }, - None => return Err(HandleError{err: "Failed to find corresponding channel", msg: None}), //TODO: panic? + None => return Err(HandleError{err: "Failed to find corresponding channel", action: None}), //TODO: panic? }; match claimable_htlcs_entry { @@ -1490,11 +1544,11 @@ impl ChannelMessageHandler for ChannelManager { match channel_state.by_id.get_mut(&msg.channel_id) { Some(chan) => { if chan.get_their_node_id() != *their_node_id { - return Err(HandleError{err: "Got a message for a channel from the wrong node!", msg: None}) + return Err(HandleError{err: "Got a message for a channel from the wrong node!", action: None}) } chan.update_fulfill_htlc(&msg)? }, - None => return Err(HandleError{err: "Failed to find corresponding channel", msg: None}) + None => return Err(HandleError{err: "Failed to find corresponding channel", action: None}) } }; if let Err(_e) = self.monitor.add_update_monitor(monitor.get_funding_txo().unwrap(), monitor) { @@ -1508,11 +1562,11 @@ impl ChannelMessageHandler for ChannelManager { let payment_hash = match channel_state.by_id.get_mut(&msg.channel_id) { Some(chan) => { if chan.get_their_node_id() != *their_node_id { - return Err(HandleError{err: "Got a message for a channel from the wrong node!", msg: None}) + return Err(HandleError{err: "Got a message for a channel from the wrong node!", action: None}) } chan.update_fail_htlc(&msg, HTLCFailReason::ErrorPacket { err: msg.reason.clone() }) }, - None => return Err(HandleError{err: "Failed to find corresponding channel", msg: None}) + None => return Err(HandleError{err: "Failed to find corresponding channel", action: None}) }?; if let Some(pending_htlc) = channel_state.claimable_htlcs.get(&payment_hash) { @@ -1538,7 +1592,7 @@ impl ChannelMessageHandler for ChannelManager { let mut hmac = Hmac::new(Sha256::new(), &um); hmac.input(&err_packet.encode()[32..]); - let mut calc_tag = [0u8; 32]; + let mut calc_tag = [0u8; 32]; hmac.raw_result(&mut calc_tag); if crypto::util::fixed_time_eq(&calc_tag, &err_packet.hmac) { const UNKNOWN_CHAN: u16 = 0x4000|10; @@ -1583,11 +1637,11 @@ impl ChannelMessageHandler for ChannelManager { match channel_state.by_id.get_mut(&msg.channel_id) { Some(chan) => { if chan.get_their_node_id() != *their_node_id { - return Err(HandleError{err: "Got a message for a channel from the wrong node!", msg: None}) + return Err(HandleError{err: "Got a message for a channel from the wrong node!", action: None}) } chan.update_fail_malformed_htlc(&msg, HTLCFailReason::Reason { failure_code: msg.failure_code, data: Vec::new() }) }, - None => return Err(HandleError{err: "Failed to find corresponding channel", msg: None}) + None => return Err(HandleError{err: "Failed to find corresponding channel", action: None}) } } @@ -1597,11 +1651,11 @@ impl ChannelMessageHandler for ChannelManager { match channel_state.by_id.get_mut(&msg.channel_id) { Some(chan) => { if chan.get_their_node_id() != *their_node_id { - return Err(HandleError{err: "Got a message for a channel from the wrong node!", msg: None}) + return Err(HandleError{err: "Got a message for a channel from the wrong node!", action: None}) } chan.commitment_signed(&msg)? }, - None => return Err(HandleError{err: "Failed to find corresponding channel", msg: None}) + None => return Err(HandleError{err: "Failed to find corresponding channel", action: None}) } }; if let Err(_e) = self.monitor.add_update_monitor(chan_monitor.get_funding_txo().unwrap(), chan_monitor) { @@ -1617,11 +1671,11 @@ impl ChannelMessageHandler for ChannelManager { match channel_state.by_id.get_mut(&msg.channel_id) { Some(chan) => { if chan.get_their_node_id() != *their_node_id { - return Err(HandleError{err: "Got a message for a channel from the wrong node!", msg: None}) + return Err(HandleError{err: "Got a message for a channel from the wrong node!", action: None}) } chan.revoke_and_ack(&msg)? }, - None => return Err(HandleError{err: "Failed to find corresponding channel", msg: None}) + None => return Err(HandleError{err: "Failed to find corresponding channel", action: None}) } }; if let Err(_e) = self.monitor.add_update_monitor(chan_monitor.get_funding_txo().unwrap(), chan_monitor) { @@ -1667,11 +1721,11 @@ impl ChannelMessageHandler for ChannelManager { match channel_state.by_id.get_mut(&msg.channel_id) { Some(chan) => { if chan.get_their_node_id() != *their_node_id { - return Err(HandleError{err: "Got a message for a channel from the wrong node!", msg: None}) + return Err(HandleError{err: "Got a message for a channel from the wrong node!", action: None}) } chan.update_fee(&*self.fee_estimator, &msg) }, - None => return Err(HandleError{err: "Failed to find corresponding channel", msg: None}) + None => return Err(HandleError{err: "Failed to find corresponding channel", action: None}) } } @@ -1681,10 +1735,10 @@ impl ChannelMessageHandler for ChannelManager { match channel_state.by_id.get_mut(&msg.channel_id) { Some(chan) => { if chan.get_their_node_id() != *their_node_id { - return Err(HandleError{err: "Got a message for a channel from the wrong node!", msg: None}) + return Err(HandleError{err: "Got a message for a channel from the wrong node!", action: None}) } if !chan.is_usable() { - return Err(HandleError{err: "Got an announcement_signatures before we were ready for it", msg: None }); + return Err(HandleError{err: "Got an announcement_signatures before we were ready for it", action: None }); } let our_node_id = self.get_our_node_id(); @@ -1705,7 +1759,7 @@ impl ChannelMessageHandler for ChannelManager { contents: announcement, }, self.get_channel_update(chan).unwrap()) // can only fail if we're not in a ready state }, - None => return Err(HandleError{err: "Failed to find corresponding channel", msg: None}) + None => return Err(HandleError{err: "Failed to find corresponding channel", action: None}) } }; let mut pending_events = self.pending_events.lock().unwrap(); @@ -1714,39 +1768,54 @@ impl ChannelMessageHandler for ChannelManager { } fn peer_disconnected(&self, their_node_id: &PublicKey, no_connection_possible: bool) { - let mut channel_state_lock = self.channel_state.lock().unwrap(); - let channel_state = channel_state_lock.borrow_parts(); - let short_to_id = channel_state.short_to_id; - if no_connection_possible { - channel_state.by_id.retain(move |_, chan| { - if chan.get_their_node_id() == *their_node_id { - if let Some(short_id) = chan.get_short_channel_id() { - short_to_id.remove(&short_id); + let mut new_events = Vec::new(); + { + let mut channel_state_lock = self.channel_state.lock().unwrap(); + let channel_state = channel_state_lock.borrow_parts(); + let short_to_id = channel_state.short_to_id; + if no_connection_possible { + channel_state.by_id.retain(|_, chan| { + if chan.get_their_node_id() == *their_node_id { + if let Some(short_id) = chan.get_short_channel_id() { + short_to_id.remove(&short_id); + } + let txn_to_broadcast = chan.force_shutdown(); + for tx in txn_to_broadcast { + self.tx_broadcaster.broadcast_transaction(&tx); + } + if let Ok(update) = self.get_channel_update(&chan) { + new_events.push(events::Event::BroadcastChannelUpdate { + msg: update + }); + } + false + } else { + true } - let txn_to_broadcast = chan.force_shutdown(); - for tx in txn_to_broadcast { - self.tx_broadcaster.broadcast_transaction(&tx); + }); + } else { + for chan in channel_state.by_id { + if chan.1.get_their_node_id() == *their_node_id { + //TODO: mark channel disabled (and maybe announce such after a timeout). Also + //fail and wipe any uncommitted outbound HTLCs as those are considered after + //reconnect. } - false - } else { - true - } - }); - } else { - for chan in channel_state.by_id { - if chan.1.get_their_node_id() == *their_node_id { - //TODO: mark channel disabled (and maybe announce such after a timeout). Also - //fail and wipe any uncommitted outbound HTLCs as those are considered after - //reconnect. } } } + if !new_events.is_empty() { + let mut pending_events = self.pending_events.lock().unwrap(); + for event in new_events.drain(..) { + pending_events.push(event); + } + } } } #[cfg(test)] mod tests { use chain::chaininterface; + use chain::transaction::OutPoint; use ln::channelmanager::{ChannelManager,OnionKeys}; use ln::router::{Route, RouteHop, Router}; use ln::msgs; @@ -1948,17 +2017,27 @@ mod tests { static mut CHAN_COUNT: u32 = 0; fn create_chan_between_nodes(node_a: &Node, node_b: &Node) -> (msgs::ChannelAnnouncement, msgs::ChannelUpdate, msgs::ChannelUpdate, Uint256, Transaction) { - let open_chan = node_a.node.create_channel(node_b.node.get_our_node_id(), 100000, 42).unwrap(); - let accept_chan = node_b.node.handle_open_channel(&node_a.node.get_our_node_id(), &open_chan).unwrap(); + node_a.node.create_channel(node_b.node.get_our_node_id(), 100000, 42).unwrap(); + + let events_1 = node_a.node.get_and_clear_pending_events(); + assert_eq!(events_1.len(), 1); + let accept_chan = match events_1[0] { + Event::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(); let chan_id = unsafe { CHAN_COUNT }; let tx; let funding_output; - let events_1 = node_a.node.get_and_clear_pending_events(); - assert_eq!(events_1.len(), 1); - match events_1[0] { + let events_2 = node_a.node.get_and_clear_pending_events(); + assert_eq!(events_2.len(), 1); + match events_2[0] { Event::FundingGenerationReady { ref temporary_channel_id, ref channel_value_satoshis, ref output_script, user_channel_id } => { assert_eq!(*channel_value_satoshis, 100000); assert_eq!(user_channel_id, 42); @@ -1966,9 +2045,9 @@ mod tests { tx = Transaction { version: chan_id as u32, lock_time: 0, input: Vec::new(), output: vec![TxOut { value: *channel_value_satoshis, script_pubkey: output_script.clone(), }]}; - funding_output = (Sha256dHash::from_data(&serialize(&tx).unwrap()[..]), 0); + funding_output = OutPoint::new(Sha256dHash::from_data(&serialize(&tx).unwrap()[..]), 0); - node_a.node.funding_transaction_generated(&temporary_channel_id, funding_output.clone()); + node_a.node.funding_transaction_generated(&temporary_channel_id, funding_output); let mut added_monitors = node_a.chan_monitor.added_monitors.lock().unwrap(); assert_eq!(added_monitors.len(), 1); assert_eq!(added_monitors[0].0, funding_output); @@ -1977,9 +2056,9 @@ mod tests { _ => panic!("Unexpected event"), } - let events_2 = node_a.node.get_and_clear_pending_events(); - assert_eq!(events_2.len(), 1); - let funding_signed = match events_2[0] { + let events_3 = node_a.node.get_and_clear_pending_events(); + assert_eq!(events_3.len(), 1); + let funding_signed = match events_3[0] { Event::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(); @@ -2000,9 +2079,9 @@ mod tests { added_monitors.clear(); } - let events_3 = node_a.node.get_and_clear_pending_events(); - assert_eq!(events_3.len(), 1); - match events_3[0] { + let events_4 = node_a.node.get_and_clear_pending_events(); + assert_eq!(events_4.len(), 1); + match events_4[0] { Event::FundingBroadcastSafe { ref funding_txo, user_channel_id } => { assert_eq!(user_channel_id, 42); assert_eq!(*funding_txo, funding_output); @@ -2011,9 +2090,9 @@ mod tests { }; confirm_transaction(&node_a.chain_monitor, &tx, chan_id); - let events_4 = node_a.node.get_and_clear_pending_events(); - assert_eq!(events_4.len(), 1); - match events_4[0] { + let events_5 = node_a.node.get_and_clear_pending_events(); + assert_eq!(events_5.len(), 1); + match events_5[0] { Event::SendFundingLocked { ref node_id, ref msg, ref announcement_sigs } => { assert_eq!(*node_id, node_b.node.get_our_node_id()); assert!(announcement_sigs.is_none()); @@ -2025,9 +2104,9 @@ mod tests { let channel_id; confirm_transaction(&node_b.chain_monitor, &tx, chan_id); - let events_5 = node_b.node.get_and_clear_pending_events(); - assert_eq!(events_5.len(), 1); - let as_announcement_sigs = match events_5[0] { + let events_6 = node_b.node.get_and_clear_pending_events(); + assert_eq!(events_6.len(), 1); + let as_announcement_sigs = match events_6[0] { Event::SendFundingLocked { ref node_id, ref msg, ref announcement_sigs } => { assert_eq!(*node_id, node_a.node.get_our_node_id()); channel_id = msg.channel_id.clone(); @@ -2038,9 +2117,9 @@ mod tests { _ => panic!("Unexpected event"), }; - let events_6 = node_a.node.get_and_clear_pending_events(); - assert_eq!(events_6.len(), 1); - let (announcement, as_update) = match events_6[0] { + let events_7 = node_a.node.get_and_clear_pending_events(); + assert_eq!(events_7.len(), 1); + let (announcement, as_update) = match events_7[0] { Event::BroadcastChannelAnnouncement { ref msg, ref update_msg } => { (msg, update_msg) }, @@ -2048,9 +2127,9 @@ mod tests { }; node_b.node.handle_announcement_signatures(&node_a.node.get_our_node_id(), &as_announcement_sigs).unwrap(); - let events_7 = node_b.node.get_and_clear_pending_events(); - assert_eq!(events_7.len(), 1); - let bs_update = match events_7[0] { + let events_8 = node_b.node.get_and_clear_pending_events(); + assert_eq!(events_8.len(), 1); + let bs_update = match events_8[0] { Event::BroadcastChannelAnnouncement { ref msg, ref update_msg } => { assert!(*announcement == *msg); update_msg @@ -2075,7 +2154,7 @@ mod tests { (chan_announcement.1, chan_announcement.2, chan_announcement.3, chan_announcement.4) } - fn close_channel(outbound_node: &Node, inbound_node: &Node, channel_id: &Uint256, funding_tx: Transaction, close_inbound_first: bool) { + fn close_channel(outbound_node: &Node, inbound_node: &Node, channel_id: &Uint256, 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_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); @@ -2111,6 +2190,26 @@ mod tests { let mut funding_tx_map = HashMap::new(); funding_tx_map.insert(funding_tx.txid(), funding_tx); tx_a.verify(&funding_tx_map).unwrap(); + + let events_1 = node_a.get_and_clear_pending_events(); + assert_eq!(events_1.len(), 1); + let as_update = match events_1[0] { + Event::BroadcastChannelUpdate { ref msg } => { + msg.clone() + }, + _ => panic!("Unexpected event"), + }; + + let events_2 = node_b.get_and_clear_pending_events(); + assert_eq!(events_2.len(), 1); + let bs_update = match events_2[0] { + Event::BroadcastChannelUpdate { ref msg } => { + msg.clone() + }, + _ => panic!("Unexpected event"), + }; + + (as_update, bs_update) } struct SendEvent { @@ -2142,17 +2241,16 @@ mod tests { }; let mut payment_event = { - let msgs = origin_node.node.send_payment(route, our_payment_hash).unwrap().unwrap(); + origin_node.node.send_payment(route, our_payment_hash).unwrap(); { let mut added_monitors = origin_node.chan_monitor.added_monitors.lock().unwrap(); assert_eq!(added_monitors.len(), 1); added_monitors.clear(); } - SendEvent { - node_id: expected_route[0].node.get_our_node_id(), - msgs: vec!(msgs.0), - commitment_msg: msgs.1, - } + + let mut events = origin_node.node.get_and_clear_pending_events(); + assert_eq!(events.len(), 1); + SendEvent::from_event(events.remove(0)) }; let mut prev_node = origin_node; @@ -2212,9 +2310,7 @@ mod tests { assert_eq!(added_monitors.len(), 1); added_monitors.clear(); } - for event in events_2.drain(..) { - payment_event = SendEvent::from_event(event); - } + payment_event = SendEvent::from_event(events_2.remove(0)); assert_eq!(payment_event.msgs.len(), 1); } @@ -2637,6 +2733,31 @@ mod tests { res } + fn get_announce_close_broadcast_events(nodes: &Vec, a: usize, b: usize) { + let events_1 = nodes[a].node.get_and_clear_pending_events(); + assert_eq!(events_1.len(), 1); + let as_update = match events_1[0] { + Event::BroadcastChannelUpdate { ref msg } => { + msg.clone() + }, + _ => panic!("Unexpected event"), + }; + + let events_2 = nodes[b].node.get_and_clear_pending_events(); + assert_eq!(events_2.len(), 1); + let bs_update = match events_2[0] { + Event::BroadcastChannelUpdate { ref msg } => { + msg.clone() + }, + _ => panic!("Unexpected event"), + }; + + for node in nodes { + node.router.handle_channel_update(&as_update).unwrap(); + node.router.handle_channel_update(&bs_update).unwrap(); + } + } + #[test] fn channel_monitor_network_test() { // Simple test which builds a network of ChannelManagers, connects them to each other, and @@ -2663,6 +2784,7 @@ mod tests { nodes[0].chain_monitor.block_connected_checked(&header, 1, &[&node_txn[0]; 1], &[4; 1]); assert_eq!(nodes[0].tx_broadcaster.txn_broadcasted.lock().unwrap().len(), 0); } + get_announce_close_broadcast_events(&nodes, 0, 1); assert_eq!(nodes[0].node.list_channels().len(), 0); assert_eq!(nodes[1].node.list_channels().len(), 1); @@ -2677,6 +2799,7 @@ mod tests { nodes[2].chain_monitor.block_connected_checked(&header, 1, &[&node_txn[0]; 1], &[4; 1]); assert_eq!(nodes[2].tx_broadcaster.txn_broadcasted.lock().unwrap().len(), 0); } + get_announce_close_broadcast_events(&nodes, 1, 2); assert_eq!(nodes[1].node.list_channels().len(), 0); assert_eq!(nodes[2].node.list_channels().len(), 1); @@ -2716,6 +2839,7 @@ mod tests { check_preimage_claim(&nodes[3], &node_txn); } + get_announce_close_broadcast_events(&nodes, 2, 3); assert_eq!(nodes[2].node.list_channels().len(), 0); assert_eq!(nodes[3].node.list_channels().len(), 1); @@ -2749,12 +2873,12 @@ mod tests { check_preimage_claim(&nodes[4], &node_txn); } + get_announce_close_broadcast_events(&nodes, 3, 4); assert_eq!(nodes[3].node.list_channels().len(), 0); assert_eq!(nodes[4].node.list_channels().len(), 0); - // TODO: Need to reenable this when we fix local route tracking // Create some new channels: - /*let chan_5 = create_announced_chan_between_nodes(&nodes, 0, 1); + let chan_5 = create_announced_chan_between_nodes(&nodes, 0, 1); // A pending HTLC which will be revoked: let payment_preimage_3 = route_payment(&nodes[0], &vec!(&nodes[1])[..], 3000000).0; @@ -2787,7 +2911,7 @@ mod tests { } get_announce_close_broadcast_events(&nodes, 0, 1); assert_eq!(nodes[0].node.list_channels().len(), 0); - assert_eq!(nodes[1].node.list_channels().len(), 0);*/ + assert_eq!(nodes[1].node.list_channels().len(), 0); // Check that we processed all pending events for node in nodes {