X-Git-Url: http://git.bitcoin.ninja/index.cgi?a=blobdiff_plain;f=src%2Fln%2Fchannelmanager.rs;h=a957a96df73be073d6b4e9fd32aabdd8f80b2b74;hb=5bb7ba5b7d7f81977705f0ed5652c9924c7a1d56;hp=ae5842463b9c843531d4aca089eaf334d3459e61;hpb=4ccb1e493201719493eeb1b081a4d1fd6e8038b9;p=rust-lightning diff --git a/src/ln/channelmanager.rs b/src/ln/channelmanager.rs index ae584246..a957a96d 100644 --- a/src/ln/channelmanager.rs +++ b/src/ln/channelmanager.rs @@ -12,11 +12,12 @@ use bitcoin::blockdata::block::BlockHeader; use bitcoin::blockdata::transaction::Transaction; use bitcoin::blockdata::constants::genesis_block; use bitcoin::network::constants::Network; -use bitcoin::util::hash::{BitcoinHash, Sha256dHash}; +use bitcoin::util::hash::BitcoinHash; use bitcoin_hashes::{Hash, HashEngine}; use bitcoin_hashes::hmac::{Hmac, HmacEngine}; use bitcoin_hashes::sha256::Hash as Sha256; +use bitcoin_hashes::sha256d::Hash as Sha256dHash; use bitcoin_hashes::cmp::fixed_time_eq; use secp256k1::key::{SecretKey,PublicKey}; @@ -27,14 +28,15 @@ use secp256k1; use chain::chaininterface::{BroadcasterInterface,ChainListener,ChainWatchInterface,FeeEstimator}; use chain::transaction::OutPoint; use ln::channel::{Channel, ChannelError}; -use ln::channelmonitor::{ChannelMonitor, ChannelMonitorUpdateErr, ManyChannelMonitor, CLTV_CLAIM_BUFFER, HTLC_FAIL_TIMEOUT_BLOCKS, HTLC_FAIL_ANTI_REORG_DELAY}; +use ln::channelmonitor::{ChannelMonitor, ChannelMonitorUpdateErr, ManyChannelMonitor, CLTV_CLAIM_BUFFER, LATENCY_GRACE_PERIOD_BLOCKS, ANTI_REORG_DELAY}; use ln::router::Route; use ln::msgs; +use ln::msgs::LocalFeatures; use ln::onion_utils; use ln::msgs::{ChannelMessageHandler, DecodeError, HandleError}; use chain::keysinterface::KeysInterface; use util::config::UserConfig; -use util::{byte_utils, events, rng}; +use util::{byte_utils, events}; use util::ser::{Readable, ReadableArgs, Writeable, Writer}; use util::chacha20::ChaCha20; use util::logger::Logger; @@ -45,7 +47,7 @@ use std::collections::{HashMap, hash_map, HashSet}; use std::io::Cursor; use std::sync::{Arc, Mutex, MutexGuard, RwLock}; use std::sync::atomic::{AtomicUsize, Ordering}; -use std::time::{Instant,Duration}; +use std::time::Duration; // We hold various information about HTLC relay in the HTLC objects in Channel itself: // @@ -206,17 +208,26 @@ impl MsgHandleErrInternal { }, }), }, + ChannelError::CloseDelayBroadcast { msg, .. } => HandleError { + err: msg, + action: Some(msgs::ErrorAction::SendErrorMessage { + msg: msgs::ErrorMessage { + channel_id, + data: msg.to_string() + }, + }), + }, }, shutdown_finish: None, } } } -/// We hold back HTLCs we intend to relay for a random interval in the range (this, 5*this). This -/// provides some limited amount of privacy. Ideally this would range from somewhere like 1 second -/// to 30 seconds, but people expect lightning to be, you know, kinda fast, sadly. We could -/// probably increase this significantly. -const MIN_HTLC_RELAY_HOLDING_CELL_MILLIS: u32 = 50; +/// We hold back HTLCs we intend to relay for a random interval greater than this (see +/// Event::PendingHTLCsForwardable for the API guidelines indicating how long should be waited). +/// This provides some limited amount of privacy. Ideally this would range from somewhere like one +/// second to 30 seconds, but people expect lightning to be, you know, kinda fast, sadly. +const MIN_HTLC_RELAY_HOLDING_CELL_MILLIS: u64 = 100; pub(super) enum HTLCForwardInfo { AddHTLC { @@ -246,16 +257,17 @@ pub(super) enum RAACommitmentOrder { pub(super) struct ChannelHolder { pub(super) by_id: HashMap<[u8; 32], Channel>, pub(super) short_to_id: HashMap, - pub(super) next_forward: Instant, /// short channel id -> forward infos. Key of 0 means payments received /// Note that while this is held in the same mutex as the channels themselves, no consistency /// guarantees are made about the existence of a channel with the short id here, nor the short /// ids in the PendingForwardHTLCInfo! pub(super) forward_htlcs: HashMap>, + /// payment_hash -> Vec<(amount_received, htlc_source)> for tracking things that were to us and + /// can be failed/claimed by the user /// Note that while this is held in the same mutex as the channels themselves, no consistency /// guarantees are made about the channels given here actually existing anymore by the time you /// go to read them! - pub(super) claimable_htlcs: HashMap>, + pub(super) claimable_htlcs: HashMap>, /// Messages to send to peers - pushed to in the same lock that they are generated in (except /// for broadcast messages, where ordering isn't as strict). pub(super) pending_msg_events: Vec, @@ -263,9 +275,8 @@ pub(super) struct ChannelHolder { pub(super) struct MutChannelHolder<'a> { pub(super) by_id: &'a mut HashMap<[u8; 32], Channel>, pub(super) short_to_id: &'a mut HashMap, - pub(super) next_forward: &'a mut Instant, pub(super) forward_htlcs: &'a mut HashMap>, - pub(super) claimable_htlcs: &'a mut HashMap>, + pub(super) claimable_htlcs: &'a mut HashMap>, pub(super) pending_msg_events: &'a mut Vec, } impl ChannelHolder { @@ -273,7 +284,6 @@ impl ChannelHolder { MutChannelHolder { by_id: &mut self.by_id, short_to_id: &mut self.short_to_id, - next_forward: &mut self.next_forward, forward_htlcs: &mut self.forward_htlcs, claimable_htlcs: &mut self.claimable_htlcs, pending_msg_events: &mut self.pending_msg_events, @@ -340,6 +350,12 @@ pub struct ChannelManager { logger: Arc, } +/// The amount of time we require our counterparty wait to claim their money (ie time between when +/// we, or our watchtower, must check for them having broadcast a theft transaction). +pub(crate) const BREAKDOWN_TIMEOUT: u16 = 6 * 24; +/// The amount of time we're willing to wait to claim money back to us +pub(crate) const MAX_LOCAL_BREAKDOWN_TIMEOUT: u16 = 6 * 24 * 7; + /// The minimum number of blocks between an inbound HTLC's CLTV and the corresponding outbound /// HTLC's CLTV. This should always be a few blocks greater than channelmonitor::CLTV_CLAIM_BUFFER, /// ie the node we forwarded the payment on to should always have enough room to reliably time out @@ -348,20 +364,21 @@ pub struct ChannelManager { const CLTV_EXPIRY_DELTA: u16 = 6 * 12; //TODO? pub(super) const CLTV_FAR_FAR_AWAY: u32 = 6 * 24 * 7; //TODO? -// Check that our CLTV_EXPIRY is at least CLTV_CLAIM_BUFFER + 2*HTLC_FAIL_TIMEOUT_BLOCKS + -// HTLC_FAIL_ANTI_REORG_DELAY, ie that if the next-hop peer fails the HTLC within -// HTLC_FAIL_TIMEOUT_BLOCKS then we'll still have HTLC_FAIL_TIMEOUT_BLOCKS left to fail it -// backwards ourselves before hitting the CLTV_CLAIM_BUFFER point and failing the channel -// on-chain to time out the HTLC. +// Check that our CLTV_EXPIRY is at least CLTV_CLAIM_BUFFER + ANTI_REORG_DELAY + LATENCY_GRACE_PERIOD_BLOCKS, +// ie that if the next-hop peer fails the HTLC within +// LATENCY_GRACE_PERIOD_BLOCKS then we'll still have CLTV_CLAIM_BUFFER left to timeout it onchain, +// then waiting ANTI_REORG_DELAY to be reorg-safe on the outbound HLTC and +// failing the corresponding htlc backward, and us now seeing the last block of ANTI_REORG_DELAY before +// LATENCY_GRACE_PERIOD_BLOCKS. #[deny(const_err)] #[allow(dead_code)] -const CHECK_CLTV_EXPIRY_SANITY: u32 = CLTV_EXPIRY_DELTA as u32 - 2*HTLC_FAIL_TIMEOUT_BLOCKS - CLTV_CLAIM_BUFFER - HTLC_FAIL_ANTI_REORG_DELAY; +const CHECK_CLTV_EXPIRY_SANITY: u32 = CLTV_EXPIRY_DELTA as u32 - LATENCY_GRACE_PERIOD_BLOCKS - CLTV_CLAIM_BUFFER - ANTI_REORG_DELAY - LATENCY_GRACE_PERIOD_BLOCKS; // Check for ability of an attacker to make us fail on-chain by delaying inbound claim. See // ChannelMontior::would_broadcast_at_height for a description of why this is needed. #[deny(const_err)] #[allow(dead_code)] -const CHECK_CLTV_EXPIRY_SANITY_2: u32 = CLTV_EXPIRY_DELTA as u32 - HTLC_FAIL_TIMEOUT_BLOCKS - 2*CLTV_CLAIM_BUFFER; +const CHECK_CLTV_EXPIRY_SANITY_2: u32 = CLTV_EXPIRY_DELTA as u32 - LATENCY_GRACE_PERIOD_BLOCKS - 2*CLTV_CLAIM_BUFFER; macro_rules! secp_call { ( $res: expr, $err: expr ) => { @@ -388,6 +405,20 @@ pub struct ChannelDetails { pub channel_value_satoshis: u64, /// The user_id passed in to create_channel, or 0 if the channel was inbound. pub user_id: u64, + /// The available outbound capacity for sending HTLCs to the remote peer. This does not include + /// any pending HTLCs which are not yet fully resolved (and, thus, who's balance is not + /// available for inclusion in new outbound HTLCs). This further does not include any pending + /// outgoing HTLCs which are awaiting some other resolution to be sent. + pub outbound_capacity_msat: u64, + /// The available inbound capacity for the remote peer to send HTLCs to us. This does not + /// include any pending HTLCs which are not yet fully resolved (and, thus, who's balance is not + /// available for inclusion in new inbound HTLCs). + /// Note that there are some corner cases not fully handled here, so the actual available + /// inbound capacity may be slightly higher than this. + pub inbound_capacity_msat: u64, + /// True if the channel is (a) confirmed and funding_locked messages have been exchanged, (b) + /// the peer is connected, and (c) no monitor update failure is pending resolution. + pub is_live: bool, } macro_rules! handle_error { @@ -425,6 +456,7 @@ macro_rules! break_chan_entry { } break Err(MsgHandleErrInternal::from_finish_shutdown(msg, channel_id, chan.force_shutdown(), $self.get_channel_update(&chan).ok())) }, + Err(ChannelError::CloseDelayBroadcast { .. }) => { panic!("Wait is only generated on receipt of channel_reestablish, which is handled by try_chan_entry, we don't bother to support it here"); } } } } @@ -444,6 +476,31 @@ macro_rules! try_chan_entry { } return Err(MsgHandleErrInternal::from_finish_shutdown(msg, channel_id, chan.force_shutdown(), $self.get_channel_update(&chan).ok())) }, + Err(ChannelError::CloseDelayBroadcast { msg, update }) => { + log_error!($self, "Channel {} need to be shutdown but closing transactions not broadcast due to {}", log_bytes!($entry.key()[..]), msg); + let (channel_id, mut chan) = $entry.remove_entry(); + if let Some(short_id) = chan.get_short_channel_id() { + $channel_state.short_to_id.remove(&short_id); + } + if let Some(update) = update { + if let Err(e) = $self.monitor.add_update_monitor(update.get_funding_txo().unwrap(), update) { + match e { + // Upstream channel is dead, but we want at least to fail backward HTLCs to save + // downstream channels. In case of PermanentFailure, we are not going to be able + // to claim back to_remote output on remote commitment transaction. Doesn't + // make a difference here, we are concern about HTLCs circuit, not onchain funds. + ChannelMonitorUpdateErr::PermanentFailure => {}, + ChannelMonitorUpdateErr::TemporaryFailure => {}, + } + } + } + let mut shutdown_res = chan.force_shutdown(); + if shutdown_res.0.len() >= 1 { + log_error!($self, "You have a toxic local commitment transaction {} avaible in channel monitor, read comment in ChannelMonitor::get_latest_local_commitment_txn to be informed of manual action to take", shutdown_res.0[0].txid()); + } + shutdown_res.0.clear(); + return Err(MsgHandleErrInternal::from_finish_shutdown(msg, channel_id, shutdown_res, $self.get_channel_update(&chan).ok())) + } } } } @@ -455,6 +512,7 @@ macro_rules! handle_monitor_err { ($self: ident, $err: expr, $channel_state: expr, $entry: expr, $action_type: path, $resend_raa: expr, $resend_commitment: expr, $failed_forwards: expr, $failed_fails: expr) => { match $err { ChannelMonitorUpdateErr::PermanentFailure => { + log_error!($self, "Closing channel {} due to monitor update PermanentFailure", log_bytes!($entry.key()[..])); let (channel_id, mut chan) = $entry.remove_entry(); if let Some(short_id) = chan.get_short_channel_id() { $channel_state.short_to_id.remove(&short_id); @@ -472,13 +530,25 @@ macro_rules! handle_monitor_err { res }, ChannelMonitorUpdateErr::TemporaryFailure => { + log_info!($self, "Disabling channel {} due to monitor update TemporaryFailure. On restore will send {} and process {} forwards and {} fails", + log_bytes!($entry.key()[..]), + if $resend_commitment && $resend_raa { + match $action_type { + RAACommitmentOrder::CommitmentFirst => { "commitment then RAA" }, + RAACommitmentOrder::RevokeAndACKFirst => { "RAA then commitment" }, + } + } else if $resend_commitment { "commitment" } + else if $resend_raa { "RAA" } + else { "nothing" }, + (&$failed_forwards as &Vec<(PendingForwardHTLCInfo, u64)>).len(), + (&$failed_fails as &Vec<(HTLCSource, PaymentHash, HTLCFailReason)>).len()); if !$resend_commitment { debug_assert!($action_type == RAACommitmentOrder::RevokeAndACKFirst || !$resend_raa); } if !$resend_raa { debug_assert!($action_type == RAACommitmentOrder::CommitmentFirst || !$resend_commitment); } - $entry.get_mut().monitor_update_failed($action_type, $resend_raa, $resend_commitment, $failed_forwards, $failed_fails); + $entry.get_mut().monitor_update_failed($resend_raa, $resend_commitment, $failed_forwards, $failed_fails); Err(MsgHandleErrInternal::from_chan_no_close(ChannelError::Ignore("Failed to update ChannelMonitor"), *$entry.key())) }, } @@ -533,7 +603,6 @@ impl ChannelManager { channel_state: Mutex::new(ChannelHolder{ by_id: HashMap::new(), short_to_id: HashMap::new(), - next_forward: Instant::now(), forward_htlcs: HashMap::new(), claimable_htlcs: HashMap::new(), pending_msg_events: Vec::new(), @@ -597,12 +666,16 @@ impl ChannelManager { let channel_state = self.channel_state.lock().unwrap(); let mut res = Vec::with_capacity(channel_state.by_id.len()); for (channel_id, channel) in channel_state.by_id.iter() { + let (inbound_capacity_msat, outbound_capacity_msat) = channel.get_inbound_outbound_available_balance_msat(); res.push(ChannelDetails { channel_id: (*channel_id).clone(), short_channel_id: channel.get_short_channel_id(), remote_network_id: channel.get_their_node_id(), channel_value_satoshis: channel.get_value_satoshis(), + inbound_capacity_msat, + outbound_capacity_msat, user_id: channel.get_user_id(), + is_live: channel.is_live(), }); } res @@ -610,6 +683,9 @@ impl ChannelManager { /// Gets the list of usable channels, in random order. Useful as an argument to /// Router::get_route to ensure non-announced channels are used. + /// + /// These are guaranteed to have their is_live value set to true, see the documentation for + /// ChannelDetails::is_live for more info on exactly what the criteria are. pub fn list_usable_channels(&self) -> Vec { let channel_state = self.channel_state.lock().unwrap(); let mut res = Vec::with_capacity(channel_state.by_id.len()); @@ -618,12 +694,16 @@ impl ChannelManager { // internal/external nomenclature, but that's ok cause that's probably what the user // really wanted anyway. if channel.is_live() { + let (inbound_capacity_msat, outbound_capacity_msat) = channel.get_inbound_outbound_available_balance_msat(); res.push(ChannelDetails { channel_id: (*channel_id).clone(), short_channel_id: channel.get_short_channel_id(), remote_network_id: channel.get_their_node_id(), channel_value_satoshis: channel.get_value_satoshis(), + inbound_capacity_msat, + outbound_capacity_msat, user_id: channel.get_user_id(), + is_live: true, }); } } @@ -804,7 +884,7 @@ impl ChannelManager { let pending_forward_info = if next_hop_data.hmac == [0; 32] { // OUR PAYMENT! // final_expiry_too_soon - if (msg.cltv_expiry as u64) < self.latest_block_height.load(Ordering::Acquire) as u64 + (CLTV_CLAIM_BUFFER + HTLC_FAIL_TIMEOUT_BLOCKS) as u64 { + if (msg.cltv_expiry as u64) < self.latest_block_height.load(Ordering::Acquire) as u64 + (CLTV_CLAIM_BUFFER + LATENCY_GRACE_PERIOD_BLOCKS) as u64 { return_err!("The final CLTV expiry is too soon to handle", 17, &[0;0]); } // final_incorrect_htlc_amount @@ -896,8 +976,8 @@ impl ChannelManager { break Some(("Forwarding node has tampered with the intended HTLC values or origin node has an obsolete cltv_expiry_delta", 0x1000 | 13, Some(self.get_channel_update(chan).unwrap()))); } let cur_height = self.latest_block_height.load(Ordering::Acquire) as u32 + 1; - // We want to have at least HTLC_FAIL_TIMEOUT_BLOCKS to fail prior to going on chain CLAIM_BUFFER blocks before expiration - if msg.cltv_expiry <= cur_height + CLTV_CLAIM_BUFFER + HTLC_FAIL_TIMEOUT_BLOCKS as u32 { // expiry_too_soon + // We want to have at least LATENCY_GRACE_PERIOD_BLOCKS to fail prior to going on chain CLAIM_BUFFER blocks before expiration + if msg.cltv_expiry <= cur_height + CLTV_CLAIM_BUFFER + LATENCY_GRACE_PERIOD_BLOCKS as u32 { // expiry_too_soon break Some(("CLTV expiry is too close", 0x1000 | 14, Some(self.get_channel_update(chan).unwrap()))); } if msg.cltv_expiry > cur_height + CLTV_FAR_FAR_AWAY as u32 { // expiry_too_far @@ -949,7 +1029,7 @@ impl ChannelManager { excess_data: Vec::new(), }; - let msg_hash = Sha256dHash::from_data(&unsigned.encode()[..]); + let msg_hash = Sha256dHash::hash(&unsigned.encode()[..]); let sig = self.secp_ctx.sign(&hash_to_message!(&msg_hash[..]), &self.our_network_key); Ok(msgs::ChannelUpdate { @@ -1086,7 +1166,7 @@ impl ChannelManager { pub fn funding_transaction_generated(&self, temporary_channel_id: &[u8; 32], funding_txo: OutPoint) { let _ = self.total_consistency_lock.read().unwrap(); - let (chan, msg, chan_monitor) = { + let (mut chan, msg, chan_monitor) = { let (res, chan) = { let mut channel_state = self.channel_state.lock().unwrap(); match channel_state.by_id.remove(temporary_channel_id) { @@ -1117,8 +1197,30 @@ impl ChannelManager { }; // Because we have exclusive ownership of the channel here we can release the channel_state // lock before add_update_monitor - if let Err(_e) = self.monitor.add_update_monitor(chan_monitor.get_funding_txo().unwrap(), chan_monitor) { - unimplemented!(); + if let Err(e) = self.monitor.add_update_monitor(chan_monitor.get_funding_txo().unwrap(), chan_monitor) { + match e { + ChannelMonitorUpdateErr::PermanentFailure => { + match handle_error!(self, Err(MsgHandleErrInternal::from_finish_shutdown("ChannelMonitor storage failure", *temporary_channel_id, chan.force_shutdown(), None))) { + Err(e) => { + log_error!(self, "Failed to store ChannelMonitor update for funding tx generation"); + let mut channel_state = self.channel_state.lock().unwrap(); + channel_state.pending_msg_events.push(events::MessageSendEvent::HandleError { + node_id: chan.get_their_node_id(), + action: e.action, + }); + return; + }, + Ok(()) => unreachable!(), + } + }, + ChannelMonitorUpdateErr::TemporaryFailure => { + // Its completely fine to continue with a FundingCreated until the monitor + // update is persisted, as long as we don't generate the FundingBroadcastSafe + // until the monitor has been safely persisted (as funding broadcast is not, + // in fact, safe). + chan.monitor_update_failed(false, false, Vec::new(), Vec::new()); + }, + } } let mut channel_state = self.channel_state.lock().unwrap(); @@ -1143,7 +1245,7 @@ impl ChannelManager { Ok(res) => res, Err(_) => return None, // Only in case of state precondition violations eg channel is closing }; - let msghash = hash_to_message!(&Sha256dHash::from_data(&announcement.encode()[..])[..]); + let msghash = hash_to_message!(&Sha256dHash::hash(&announcement.encode()[..])[..]); let our_node_sig = self.secp_ctx.sign(&msghash, &self.our_network_key); Some(msgs::AnnouncementSignatures { @@ -1168,10 +1270,6 @@ impl ChannelManager { let mut channel_state_lock = self.channel_state.lock().unwrap(); let channel_state = channel_state_lock.borrow_parts(); - if cfg!(not(feature = "fuzztarget")) && Instant::now() < *channel_state.next_forward { - return; - } - for (short_chan_id, mut pending_forwards) in channel_state.forward_htlcs.drain() { if short_chan_id != 0 { let forward_chan_id = match channel_state.short_to_id.get(&short_chan_id) { @@ -1308,8 +1406,8 @@ impl ChannelManager { incoming_packet_shared_secret: forward_info.incoming_shared_secret, }; match channel_state.claimable_htlcs.entry(forward_info.payment_hash) { - hash_map::Entry::Occupied(mut entry) => entry.get_mut().push(prev_hop_data), - hash_map::Entry::Vacant(entry) => { entry.insert(vec![prev_hop_data]); }, + hash_map::Entry::Occupied(mut entry) => entry.get_mut().push((forward_info.amt_to_forward, prev_hop_data)), + hash_map::Entry::Vacant(entry) => { entry.insert(vec![(forward_info.amt_to_forward, prev_hop_data)]); }, }; new_events.push(events::Event::PaymentReceived { payment_hash: forward_info.payment_hash, @@ -1354,20 +1452,21 @@ impl ChannelManager { } /// Indicates that the preimage for payment_hash is unknown or the received amount is incorrect - /// after a PaymentReceived event. - /// expected_value is the value you expected the payment to be for (not the amount it actually - /// was for from the PaymentReceived event). - pub fn fail_htlc_backwards(&self, payment_hash: &PaymentHash, expected_value: u64) -> bool { + /// after a PaymentReceived event, failing the HTLC back to its origin and freeing resources + /// along the path (including in our own channel on which we received it). + /// Returns false if no payment was found to fail backwards, true if the process of failing the + /// HTLC backwards has been started. + pub fn fail_htlc_backwards(&self, payment_hash: &PaymentHash) -> 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 { - for htlc_with_hash in sources.drain(..) { + for (recvd_value, htlc_with_hash) in sources.drain(..) { if channel_state.is_none() { channel_state = Some(self.channel_state.lock().unwrap()); } self.fail_htlc_backwards_internal(channel_state.take().unwrap(), HTLCSource::PreviousHopData(htlc_with_hash), payment_hash, - HTLCFailReason::Reason { failure_code: 0x4000 | 15, data: byte_utils::be64_to_array(expected_value).to_vec() }); + HTLCFailReason::Reason { failure_code: 0x4000 | 15, data: byte_utils::be64_to_array(recvd_value).to_vec() }); } true } else { false } @@ -1450,8 +1549,7 @@ impl ChannelManager { let mut forward_event = None; if channel_state_lock.forward_htlcs.is_empty() { - forward_event = Some(Instant::now() + Duration::from_millis(((rng::rand_f32() * 4.0 + 1.0) * MIN_HTLC_RELAY_HOLDING_CELL_MILLIS as f32) as u64)); - channel_state_lock.next_forward = forward_event.unwrap(); + forward_event = Some(Duration::from_millis(MIN_HTLC_RELAY_HOLDING_CELL_MILLIS)); } match channel_state_lock.forward_htlcs.entry(short_channel_id) { hash_map::Entry::Occupied(mut entry) => { @@ -1485,7 +1583,10 @@ impl ChannelManager { 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 { - for htlc_with_hash in sources.drain(..) { + // TODO: We should require the user specify the expected amount so that we can claim + // only payments for the correct amount, and reject payments for incorrect amounts + // (which are probably middle nodes probing to break our privacy). + for (_, htlc_with_hash) in sources.drain(..) { if channel_state.is_none() { channel_state = Some(self.channel_state.lock().unwrap()); } self.claim_funds_internal(channel_state.take().unwrap(), HTLCSource::PreviousHopData(htlc_with_hash), payment_preimage); } @@ -1583,6 +1684,7 @@ impl ChannelManager { let mut close_results = Vec::new(); let mut htlc_forwards = Vec::new(); let mut htlc_failures = Vec::new(); + let mut pending_events = Vec::new(); let _ = self.total_consistency_lock.read().unwrap(); { @@ -1617,7 +1719,7 @@ impl ChannelManager { ChannelMonitorUpdateErr::TemporaryFailure => true, } } else { - let (raa, commitment_update, order, pending_forwards, mut pending_failures) = channel.monitor_updating_restored(); + let (raa, commitment_update, order, pending_forwards, mut pending_failures, needs_broadcast_safe, funding_locked) = channel.monitor_updating_restored(); if !pending_forwards.is_empty() { htlc_forwards.push((channel.get_short_channel_id().expect("We can't have pending forwards before funding confirmation"), pending_forwards)); } @@ -1649,12 +1751,33 @@ impl ChannelManager { handle_cs!(); }, } + if needs_broadcast_safe { + pending_events.push(events::Event::FundingBroadcastSafe { + funding_txo: channel.get_funding_txo().unwrap(), + user_channel_id: channel.get_user_id(), + }); + } + if let Some(msg) = funding_locked { + pending_msg_events.push(events::MessageSendEvent::SendFundingLocked { + node_id: channel.get_their_node_id(), + msg, + }); + if let Some(announcement_sigs) = self.get_announcement_sigs(channel) { + pending_msg_events.push(events::MessageSendEvent::SendAnnouncementSignatures { + node_id: channel.get_their_node_id(), + msg: announcement_sigs, + }); + } + short_to_id.insert(channel.get_short_channel_id().unwrap(), channel.channel_id()); + } true } } else { true } }); } + self.pending_events.lock().unwrap().append(&mut pending_events); + for failure in htlc_failures.drain(..) { self.fail_htlc_backwards_internal(self.channel_state.lock().unwrap(), failure.0, &failure.1, failure.2); } @@ -1665,12 +1788,12 @@ impl ChannelManager { } } - fn internal_open_channel(&self, their_node_id: &PublicKey, msg: &msgs::OpenChannel) -> Result<(), MsgHandleErrInternal> { + fn internal_open_channel(&self, their_node_id: &PublicKey, their_local_features: LocalFeatures, msg: &msgs::OpenChannel) -> Result<(), MsgHandleErrInternal> { if msg.chain_hash != self.genesis_hash { return Err(MsgHandleErrInternal::send_err_msg_no_close("Unknown genesis block hash", msg.temporary_channel_id.clone())); } - let channel = Channel::new_from_req(&*self.fee_estimator, &self.keys_manager, their_node_id.clone(), msg, 0, Arc::clone(&self.logger), &self.default_configuration) + let channel = Channel::new_from_req(&*self.fee_estimator, &self.keys_manager, their_node_id.clone(), their_local_features, msg, 0, Arc::clone(&self.logger), &self.default_configuration) .map_err(|e| MsgHandleErrInternal::from_chan_no_close(e, msg.temporary_channel_id))?; let mut channel_state_lock = self.channel_state.lock().unwrap(); let channel_state = channel_state_lock.borrow_parts(); @@ -1687,7 +1810,7 @@ impl ChannelManager { Ok(()) } - fn internal_accept_channel(&self, their_node_id: &PublicKey, msg: &msgs::AcceptChannel) -> Result<(), MsgHandleErrInternal> { + fn internal_accept_channel(&self, their_node_id: &PublicKey, their_local_features: LocalFeatures, msg: &msgs::AcceptChannel) -> Result<(), MsgHandleErrInternal> { let (value, output_script, user_id) = { let mut channel_lock = self.channel_state.lock().unwrap(); let channel_state = channel_lock.borrow_parts(); @@ -1697,7 +1820,7 @@ impl ChannelManager { //TODO: see issue #153, need a consistent behavior on obnoxious behavior from random node return Err(MsgHandleErrInternal::send_err_msg_no_close("Got a message for a channel from the wrong node!", msg.temporary_channel_id)); } - try_chan_entry!(self, chan.get_mut().accept_channel(&msg, &self.default_configuration), channel_state, chan); + try_chan_entry!(self, chan.get_mut().accept_channel(&msg, &self.default_configuration, their_local_features), channel_state, chan); (chan.get().get_value_satoshis(), chan.get().get_funding_redeemscript().to_v0_p2wsh(), chan.get().get_user_id()) }, //TODO: same as above @@ -1715,7 +1838,7 @@ impl ChannelManager { } fn internal_funding_created(&self, their_node_id: &PublicKey, msg: &msgs::FundingCreated) -> Result<(), MsgHandleErrInternal> { - let ((funding_msg, monitor_update), chan) = { + let ((funding_msg, monitor_update), mut chan) = { let mut channel_lock = self.channel_state.lock().unwrap(); let channel_state = channel_lock.borrow_parts(); match channel_state.by_id.entry(msg.temporary_channel_id.clone()) { @@ -1731,8 +1854,23 @@ impl ChannelManager { }; // Because we have exclusive ownership of the channel here we can release the channel_state // lock before add_update_monitor - if let Err(_e) = self.monitor.add_update_monitor(monitor_update.get_funding_txo().unwrap(), monitor_update) { - unimplemented!(); + if let Err(e) = self.monitor.add_update_monitor(monitor_update.get_funding_txo().unwrap(), monitor_update) { + match e { + ChannelMonitorUpdateErr::PermanentFailure => { + // Note that we reply with the new channel_id in error messages if we gave up on the + // channel, not the temporary_channel_id. This is compatible with ourselves, but the + // spec is somewhat ambiguous here. Not a huge deal since we'll send error messages for + // any messages referencing a previously-closed channel anyway. + return Err(MsgHandleErrInternal::from_finish_shutdown("ChannelMonitor storage failure", funding_msg.channel_id, chan.force_shutdown(), None)); + }, + ChannelMonitorUpdateErr::TemporaryFailure => { + // There's no problem signing a counterparty's funding transaction if our monitor + // hasn't persisted to disk yet - we can't lose money on a transaction that we haven't + // accepted payment from yet. We do, however, need to wait to send our funding_locked + // until we have persisted our monitor. + chan.monitor_update_failed(false, false, Vec::new(), Vec::new()); + }, + } } let mut channel_state_lock = self.channel_state.lock().unwrap(); let channel_state = channel_state_lock.borrow_parts(); @@ -1762,8 +1900,8 @@ impl ChannelManager { return Err(MsgHandleErrInternal::send_err_msg_no_close("Got a message for a channel from the wrong node!", msg.channel_id)); } let chan_monitor = try_chan_entry!(self, chan.get_mut().funding_signed(&msg), channel_state, chan); - if let Err(_e) = self.monitor.add_update_monitor(chan_monitor.get_funding_txo().unwrap(), chan_monitor) { - unimplemented!(); + if let Err(e) = self.monitor.add_update_monitor(chan_monitor.get_funding_txo().unwrap(), chan_monitor) { + return_monitor_err!(self, e, channel_state, chan, RAACommitmentOrder::RevokeAndACKFirst, false, false); } (chan.get().get_funding_txo().unwrap(), chan.get().get_user_id()) }, @@ -2057,8 +2195,7 @@ impl ChannelManager { if !pending_forwards.is_empty() { let mut channel_state = self.channel_state.lock().unwrap(); if channel_state.forward_htlcs.is_empty() { - forward_event = Some(Instant::now() + Duration::from_millis(((rng::rand_f32() * 4.0 + 1.0) * MIN_HTLC_RELAY_HOLDING_CELL_MILLIS as f32) as u64)); - channel_state.next_forward = forward_event.unwrap(); + forward_event = Some(Duration::from_millis(MIN_HTLC_RELAY_HOLDING_CELL_MILLIS)) } for (forward_info, prev_htlc_id) in pending_forwards.drain(..) { match channel_state.forward_htlcs.entry(forward_info.short_channel_id) { @@ -2163,7 +2300,7 @@ impl ChannelManager { try_chan_entry!(self, chan.get_mut().get_channel_announcement(our_node_id.clone(), self.genesis_hash.clone()), channel_state, chan); let were_node_one = announcement.node_id_1 == our_node_id; - let msghash = hash_to_message!(&Sha256dHash::from_data(&announcement.encode()[..])[..]); + let msghash = hash_to_message!(&Sha256dHash::hash(&announcement.encode()[..])[..]); if self.secp_ctx.verify(&msghash, &msg.node_signature, if were_node_one { &announcement.node_id_2 } else { &announcement.node_id_1 }).is_err() || self.secp_ctx.verify(&msghash, &msg.bitcoin_signature, if were_node_one { &announcement.bitcoin_key_2 } else { &announcement.bitcoin_key_1 }).is_err() { try_chan_entry!(self, Err(ChannelError::Close("Bad announcement_signatures node_signature")), channel_state, chan); @@ -2454,7 +2591,7 @@ impl ChainListener for ChannelManager { } /// We force-close the channel without letting our counterparty participate in the shutdown - fn block_disconnected(&self, header: &BlockHeader) { + fn block_disconnected(&self, header: &BlockHeader, _: u32) { let _ = self.total_consistency_lock.read().unwrap(); let mut failed_channels = Vec::new(); { @@ -2489,14 +2626,14 @@ impl ChainListener for ChannelManager { impl ChannelMessageHandler for ChannelManager { //TODO: Handle errors and close channel (or so) - fn handle_open_channel(&self, their_node_id: &PublicKey, msg: &msgs::OpenChannel) -> Result<(), HandleError> { + fn handle_open_channel(&self, their_node_id: &PublicKey, their_local_features: LocalFeatures, msg: &msgs::OpenChannel) -> Result<(), HandleError> { let _ = self.total_consistency_lock.read().unwrap(); - handle_error!(self, self.internal_open_channel(their_node_id, msg)) + handle_error!(self, self.internal_open_channel(their_node_id, their_local_features, msg)) } - fn handle_accept_channel(&self, their_node_id: &PublicKey, msg: &msgs::AcceptChannel) -> Result<(), HandleError> { + fn handle_accept_channel(&self, their_node_id: &PublicKey, their_local_features: LocalFeatures, msg: &msgs::AcceptChannel) -> Result<(), HandleError> { let _ = self.total_consistency_lock.read().unwrap(); - handle_error!(self, self.internal_accept_channel(their_node_id, msg)) + handle_error!(self, self.internal_accept_channel(their_node_id, their_local_features, msg)) } fn handle_funding_created(&self, their_node_id: &PublicKey, msg: &msgs::FundingCreated) -> Result<(), HandleError> { @@ -2910,7 +3047,8 @@ impl Writeable for ChannelManager { 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 { + for &(recvd_amt, ref previous_hop) in previous_hops.iter() { + recvd_amt.write(writer)?; previous_hop.write(writer)?; } } @@ -3047,7 +3185,7 @@ impl<'a, R : ::std::io::Read> ReadableArgs> for (S 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)?); + previous_hops.push((Readable::read(reader)?, Readable::read(reader)?)); } claimable_htlcs.insert(payment_hash, previous_hops); } @@ -3066,7 +3204,6 @@ impl<'a, R : ::std::io::Read> ReadableArgs> for (S channel_state: Mutex::new(ChannelHolder { by_id, short_to_id, - next_forward: Instant::now(), forward_htlcs, claimable_htlcs, pending_msg_events: Vec::new(),