X-Git-Url: http://git.bitcoin.ninja/index.cgi?a=blobdiff_plain;f=lightning%2Fsrc%2Fln%2Fchannelmanager.rs;h=18d3372024a3354838f188f109e9c5b8bcee3899;hb=c9916437468a130718ab83baa2f346b301b674ba;hp=d3b4ab96dc8beb7ed314d2045067509a64685c1b;hpb=33810eddb56490584675b8be7849ce4b46b5f535;p=rust-lightning diff --git a/lightning/src/ln/channelmanager.rs b/lightning/src/ln/channelmanager.rs index d3b4ab96..18d33720 100644 --- a/lightning/src/ln/channelmanager.rs +++ b/lightning/src/ln/channelmanager.rs @@ -30,8 +30,8 @@ use chain::transaction::OutPoint; use ln::channel::{Channel, ChannelError}; use ln::channelmonitor::{ChannelMonitor, ChannelMonitorUpdateErr, ManyChannelMonitor, CLTV_CLAIM_BUFFER, LATENCY_GRACE_PERIOD_BLOCKS, ANTI_REORG_DELAY}; use ln::router::Route; +use ln::features::InitFeatures; use ln::msgs; -use ln::msgs::LocalFeatures; use ln::onion_utils; use ln::msgs::{ChannelMessageHandler, DecodeError, LightningError}; use chain::keysinterface::{ChannelKeys, KeysInterface}; @@ -274,23 +274,11 @@ pub(super) struct ChannelHolder { /// for broadcast messages, where ordering isn't as strict). pub(super) pending_msg_events: Vec, } -pub(super) struct MutChannelHolder<'a, ChanSigner: ChannelKeys + 'a> { - pub(super) by_id: &'a mut HashMap<[u8; 32], Channel>, - pub(super) short_to_id: &'a mut HashMap, - pub(super) forward_htlcs: &'a mut HashMap>, - pub(super) claimable_htlcs: &'a mut HashMap>, - pub(super) pending_msg_events: &'a mut Vec, -} -impl ChannelHolder { - pub(super) fn borrow_parts(&mut self) -> MutChannelHolder { - MutChannelHolder { - by_id: &mut self.by_id, - short_to_id: &mut self.short_to_id, - forward_htlcs: &mut self.forward_htlcs, - claimable_htlcs: &mut self.claimable_htlcs, - pending_msg_events: &mut self.pending_msg_events, - } - } + +/// State we hold per-peer. In the future we should put channels in here, but for now we only hold +/// the latest Init features we heard from the peer. +struct PeerState { + latest_features: InitFeatures, } #[cfg(not(any(target_pointer_width = "32", target_pointer_width = "64")))] @@ -346,6 +334,14 @@ pub struct ChannelManager { channel_state: Mutex>, our_network_key: SecretKey, + /// The bulk of our storage will eventually be here (channels and message queues and the like). + /// If we are connected to a peer we always at least have an entry here, even if no channels + /// are currently open with that peer. + /// Because adding or removing an entry is rare, we usually take an outer read lock and then + /// operate on the inner value freely. Sadly, this prevents parallel operation when opening a + /// new channel. + per_peer_state: RwLock>>, + 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. @@ -408,6 +404,10 @@ pub struct ChannelDetails { pub short_channel_id: Option, /// The node_id of our counterparty pub remote_network_id: PublicKey, + /// The Features the channel counterparty provided upon last connection. + /// Useful for routing as it is the most up-to-date copy of the counterparty's features and + /// many routing-relevant features are present in the init context. + pub counterparty_features: InitFeatures, /// The value, in satoshis, of this channel as appears in the funding output pub channel_value_satoshis: u64, /// The user_id passed in to create_channel, or 0 if the channel was inbound. @@ -429,19 +429,22 @@ pub struct ChannelDetails { } macro_rules! handle_error { - ($self: ident, $internal: expr) => { + ($self: ident, $internal: expr, $their_node_id: expr, $locked_channel_state: expr) => { match $internal { Ok(msg) => Ok(msg), Err(MsgHandleErrInternal { err, shutdown_finish }) => { if let Some((shutdown_res, update_option)) = shutdown_finish { $self.finish_force_close_channel(shutdown_res); if let Some(update) = update_option { - let mut channel_state = $self.channel_state.lock().unwrap(); - channel_state.pending_msg_events.push(events::MessageSendEvent::BroadcastChannelUpdate { + $locked_channel_state.pending_msg_events.push(events::MessageSendEvent::BroadcastChannelUpdate { msg: update }); } } + log_error!($self, "{}", err.err); + if let msgs::ErrorAction::IgnoreError = err.action { + } else { $locked_channel_state.pending_msg_events.push(events::MessageSendEvent::HandleError { node_id: $their_node_id, action: err.action.clone() }); } + // Return error in case higher-API need one Err(err) }, } @@ -625,6 +628,8 @@ impl ChannelManager { }), our_network_key: keys_manager.get_node_secret(), + per_peer_state: RwLock::new(HashMap::new()), + pending_events: Mutex::new(Vec::new()), total_consistency_lock: RwLock::new(()), @@ -675,56 +680,53 @@ impl ChannelManager { Ok(()) } - /// Gets the list of open channels, in random order. See ChannelDetail field documentation for - /// more information. - pub fn list_channels(&self) -> Vec { - 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 - } - - /// 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()); - for (channel_id, channel) in channel_state.by_id.iter() { - // Note we use is_live here instead of usable which leads to somewhat confused - // internal/external nomenclature, but that's ok cause that's probably what the user - // really wanted anyway. - if channel.is_live() { + fn list_channels_with_filter)) -> bool>(&self, f: F) -> Vec { + let mut res = Vec::new(); + { + let channel_state = self.channel_state.lock().unwrap(); + res.reserve(channel_state.by_id.len()); + for (channel_id, channel) in channel_state.by_id.iter().filter(f) { 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(), + counterparty_features: InitFeatures::empty(), channel_value_satoshis: channel.get_value_satoshis(), inbound_capacity_msat, outbound_capacity_msat, user_id: channel.get_user_id(), - is_live: true, + is_live: channel.is_live(), }); } } + let per_peer_state = self.per_peer_state.read().unwrap(); + for chan in res.iter_mut() { + if let Some(peer_state) = per_peer_state.get(&chan.remote_network_id) { + chan.counterparty_features = peer_state.lock().unwrap().latest_features.clone(); + } + } res } + /// Gets the list of open channels, in random order. See ChannelDetail field documentation for + /// more information. + pub fn list_channels(&self) -> Vec { + self.list_channels_with_filter(|_| true) + } + + /// 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 { + // Note we use is_live here instead of usable which leads to somewhat confused + // internal/external nomenclature, but that's ok cause that's probably what the user + // really wanted anyway. + self.list_channels_with_filter(|&(_, ref channel)| channel.is_live()) + } + /// Begins the process of closing a channel. After this call (plus some timeout), no new HTLCs /// will be accepted on the given channel, and after additional timeout/the closing of all /// pending HTLCs, the channel will be closed on chain. @@ -735,7 +737,7 @@ impl ChannelManager { let (mut failed_htlcs, chan_option) = { let mut channel_state_lock = self.channel_state.lock().unwrap(); - let channel_state = channel_state_lock.borrow_parts(); + let channel_state = &mut *channel_state_lock; match channel_state.by_id.entry(channel_id.clone()) { hash_map::Entry::Occupied(mut chan_entry) => { let (shutdown_msg, failed_htlcs) = chan_entry.get_mut().get_shutdown()?; @@ -792,7 +794,7 @@ impl ChannelManager { let mut chan = { let mut channel_state_lock = self.channel_state.lock().unwrap(); - let channel_state = channel_state_lock.borrow_parts(); + let channel_state = &mut *channel_state_lock; if let Some(chan) = channel_state.by_id.remove(channel_id) { if let Some(short_id) = chan.get_short_channel_id() { channel_state.short_to_id.remove(&short_id); @@ -1116,15 +1118,15 @@ impl ChannelManager { let _ = self.total_consistency_lock.read().unwrap(); + let mut channel_lock = self.channel_state.lock().unwrap(); let err: Result<(), _> = loop { - let mut channel_lock = self.channel_state.lock().unwrap(); let id = match channel_lock.short_to_id.get(&route.hops.first().unwrap().short_channel_id) { None => return Err(APIError::ChannelUnavailable{err: "No channel available with first hop!"}), Some(id) => id.clone(), }; - let channel_state = channel_lock.borrow_parts(); + let channel_state = &mut *channel_lock; if let hash_map::Entry::Occupied(mut chan) = channel_state.by_id.entry(id) { match { if chan.get().get_their_node_id() != route.hops.first().unwrap().pubkey { @@ -1167,20 +1169,9 @@ impl ChannelManager { return Ok(()); }; - match handle_error!(self, err) { + match handle_error!(self, err, route.hops.first().unwrap().pubkey, channel_lock) { Ok(_) => unreachable!(), - Err(e) => { - if let msgs::ErrorAction::IgnoreError = e.action { - } else { - log_error!(self, "Got bad keys: {}!", e.err); - let mut channel_state = self.channel_state.lock().unwrap(); - channel_state.pending_msg_events.push(events::MessageSendEvent::HandleError { - node_id: route.hops.first().unwrap().pubkey, - action: e.action, - }); - } - Err(APIError::ChannelUnavailable { err: e.err }) - }, + Err(e) => { Err(APIError::ChannelUnavailable { err: e.err }) } } } @@ -1197,32 +1188,22 @@ impl ChannelManager { let _ = self.total_consistency_lock.read().unwrap(); 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) { - Some(mut chan) => { - (chan.get_outbound_funding_created(funding_txo) - .map_err(|e| if let ChannelError::Close(msg) = e { - MsgHandleErrInternal::from_finish_shutdown(msg, chan.channel_id(), chan.force_shutdown(), None) - } else { unreachable!(); }) - , chan) - }, - None => return - } + let mut channel_state = self.channel_state.lock().unwrap(); + let (res, chan) = match channel_state.by_id.remove(temporary_channel_id) { + Some(mut chan) => { + (chan.get_outbound_funding_created(funding_txo) + .map_err(|e| if let ChannelError::Close(msg) = e { + MsgHandleErrInternal::from_finish_shutdown(msg, chan.channel_id(), chan.force_shutdown(), None) + } else { unreachable!(); }) + , chan) + }, + None => return }; - match handle_error!(self, res) { + match handle_error!(self, res, chan.get_their_node_id(), channel_state) { Ok(funding_msg) => { (chan, funding_msg.0, funding_msg.1) }, - Err(e) => { - log_error!(self, "Got bad signatures: {}!", e.err); - 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; - }, + Err(_) => { return; } } }; // Because we have exclusive ownership of the channel here we can release the channel_state @@ -1230,17 +1211,12 @@ impl ChannelManager { 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!(), + { + let mut channel_state = self.channel_state.lock().unwrap(); + match handle_error!(self, Err(MsgHandleErrInternal::from_finish_shutdown("ChannelMonitor storage failure", *temporary_channel_id, chan.force_shutdown(), None)), chan.get_their_node_id(), channel_state) { + Err(_) => { return; }, + Ok(()) => unreachable!(), + } } }, ChannelMonitorUpdateErr::TemporaryFailure => { @@ -1298,7 +1274,7 @@ impl ChannelManager { let mut handle_errors = Vec::new(); { let mut channel_state_lock = self.channel_state.lock().unwrap(); - let channel_state = channel_state_lock.borrow_parts(); + let channel_state = &mut *channel_state_lock; for (short_chan_id, mut pending_forwards) in channel_state.forward_htlcs.drain() { if short_chan_id != 0 { @@ -1417,22 +1393,9 @@ impl ChannelManager { }, 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"); } }; - match handle_error!(self, err) { + match handle_error!(self, err, their_node_id, channel_state) { Ok(_) => unreachable!(), - Err(e) => { - match e.action { - msgs::ErrorAction::IgnoreError => {}, - _ => { - log_error!(self, "Got bad keys: {}!", e.err); - let mut channel_state = self.channel_state.lock().unwrap(); - channel_state.pending_msg_events.push(events::MessageSendEvent::HandleError { - node_id: their_node_id, - action: e.action, - }); - }, - } - continue; - }, + Err(_) => { continue; }, } } }; @@ -1489,19 +1452,10 @@ impl ChannelManager { }; } - for (their_node_id, err) in handle_errors.drain(..) { - match handle_error!(self, err) { - Ok(_) => {}, - Err(e) => { - if let msgs::ErrorAction::IgnoreError = e.action { - } else { - let mut channel_state = self.channel_state.lock().unwrap(); - channel_state.pending_msg_events.push(events::MessageSendEvent::HandleError { - node_id: their_node_id, - action: e.action, - }); - } - }, + if handle_errors.len() > 0 { + let mut channel_state_lock = self.channel_state.lock().unwrap(); + for (their_node_id, err) in handle_errors.drain(..) { + let _ = handle_error!(self, err, their_node_id, channel_state_lock); } } @@ -1518,8 +1472,8 @@ impl ChannelManager { pub fn timer_chan_freshness_every_min(&self) { 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(); - for (_, chan) in channel_state.by_id { + let channel_state = &mut *channel_state_lock; + for (_, chan) in channel_state.by_id.iter_mut() { if chan.is_disabled_staged() && !chan.is_live() { if let Ok(update) = self.get_channel_update(&chan) { channel_state.pending_msg_events.push(events::MessageSendEvent::BroadcastChannelUpdate { @@ -1702,7 +1656,7 @@ impl ChannelManager { }, HTLCSource::PreviousHopData(HTLCPreviousHopData { short_channel_id, htlc_id, .. }) => { //TODO: Delay the claimed_funds relaying just like we do outbound relay! - let channel_state = channel_state_lock.borrow_parts(); + let channel_state = &mut *channel_state_lock; let chan_id = match channel_state.short_to_id.get(&short_channel_id) { Some(chan_id) => chan_id.clone(), @@ -1754,19 +1708,7 @@ impl ChannelManager { return; }; - match handle_error!(self, err) { - Ok(_) => {}, - Err(e) => { - if let msgs::ErrorAction::IgnoreError = e.action { - } else { - let mut channel_state = self.channel_state.lock().unwrap(); - channel_state.pending_msg_events.push(events::MessageSendEvent::HandleError { - node_id: their_node_id, - action: e.action, - }); - } - }, - } + let _ = handle_error!(self, err, their_node_id, channel_state_lock); } /// Gets the node_id held by this ChannelManager @@ -1786,9 +1728,9 @@ impl ChannelManager { { let mut channel_lock = self.channel_state.lock().unwrap(); - let channel_state = channel_lock.borrow_parts(); - let short_to_id = channel_state.short_to_id; - let pending_msg_events = channel_state.pending_msg_events; + let channel_state = &mut *channel_lock; + let short_to_id = &mut channel_state.short_to_id; + let pending_msg_events = &mut channel_state.pending_msg_events; channel_state.by_id.retain(|_, channel| { if channel.is_awaiting_monitor_update() { let chan_monitor = channel.channel_monitor().clone(); @@ -1885,15 +1827,15 @@ impl ChannelManager { } } - fn internal_open_channel(&self, their_node_id: &PublicKey, their_local_features: LocalFeatures, msg: &msgs::OpenChannel) -> Result<(), MsgHandleErrInternal> { + fn internal_open_channel(&self, their_node_id: &PublicKey, their_features: InitFeatures, 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(), their_local_features, 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_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(); + let channel_state = &mut *channel_state_lock; 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) => { @@ -1907,20 +1849,18 @@ impl ChannelManager { Ok(()) } - fn internal_accept_channel(&self, their_node_id: &PublicKey, their_local_features: LocalFeatures, msg: &msgs::AcceptChannel) -> Result<(), MsgHandleErrInternal> { + fn internal_accept_channel(&self, their_node_id: &PublicKey, their_features: InitFeatures, 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(); + let channel_state = &mut *channel_lock; match channel_state.by_id.entry(msg.temporary_channel_id) { hash_map::Entry::Occupied(mut chan) => { if chan.get().get_their_node_id() != *their_node_id { - //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, their_local_features), channel_state, chan); + try_chan_entry!(self, chan.get_mut().accept_channel(&msg, &self.default_configuration, their_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 hash_map::Entry::Vacant(_) => return Err(MsgHandleErrInternal::send_err_msg_no_close("Failed to find corresponding channel", msg.temporary_channel_id)) } }; @@ -1937,11 +1877,10 @@ impl ChannelManager { fn internal_funding_created(&self, their_node_id: &PublicKey, msg: &msgs::FundingCreated) -> Result<(), MsgHandleErrInternal> { let ((funding_msg, monitor_update), mut chan) = { let mut channel_lock = self.channel_state.lock().unwrap(); - let channel_state = channel_lock.borrow_parts(); + let channel_state = &mut *channel_lock; match channel_state.by_id.entry(msg.temporary_channel_id.clone()) { hash_map::Entry::Occupied(mut chan) => { if chan.get().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.temporary_channel_id)); } (try_chan_entry!(self, chan.get_mut().funding_created(msg), channel_state, chan), chan.remove()) @@ -1970,7 +1909,7 @@ impl ChannelManager { } } let mut channel_state_lock = self.channel_state.lock().unwrap(); - let channel_state = channel_state_lock.borrow_parts(); + let channel_state = &mut *channel_state_lock; 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)) @@ -1989,11 +1928,10 @@ impl ChannelManager { fn internal_funding_signed(&self, their_node_id: &PublicKey, msg: &msgs::FundingSigned) -> Result<(), MsgHandleErrInternal> { let (funding_txo, user_id) = { let mut channel_lock = self.channel_state.lock().unwrap(); - let channel_state = channel_lock.borrow_parts(); + let channel_state = &mut *channel_lock; match channel_state.by_id.entry(msg.channel_id) { hash_map::Entry::Occupied(mut chan) => { if chan.get().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 chan_monitor = try_chan_entry!(self, chan.get_mut().funding_signed(&msg), channel_state, chan); @@ -2015,11 +1953,10 @@ impl ChannelManager { 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(); + let channel_state = &mut *channel_state_lock; match channel_state.by_id.entry(msg.channel_id) { hash_map::Entry::Occupied(mut chan) => { if chan.get().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)); } try_chan_entry!(self, chan.get_mut().funding_locked(&msg), channel_state, chan); @@ -2047,12 +1984,11 @@ impl ChannelManager { 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(); + let channel_state = &mut *channel_state_lock; 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 { - //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 (shutdown, closing_signed, dropped_htlcs) = try_chan_entry!(self, chan_entry.get_mut().shutdown(&*self.fee_estimator, &msg), channel_state, chan_entry); @@ -2095,11 +2031,10 @@ impl ChannelManager { 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(); + let channel_state = &mut *channel_state_lock; 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 { - //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 (closing_signed, tx) = try_chan_entry!(self, chan_entry.get_mut().closing_signed(&*self.fee_estimator, &msg), channel_state, chan_entry); @@ -2150,12 +2085,11 @@ impl ChannelManager { //but we should prevent it anyway. let (mut pending_forward_info, mut channel_state_lock) = self.decode_update_add_htlc_onion(msg); - let channel_state = channel_state_lock.borrow_parts(); + let channel_state = &mut *channel_state_lock; match channel_state.by_id.entry(msg.channel_id) { hash_map::Entry::Occupied(mut chan) => { if chan.get().get_their_node_id() != *their_node_id { - //TODO: here MsgHandleErrInternal, #153 case return Err(MsgHandleErrInternal::send_err_msg_no_close("Got a message for a channel from the wrong node!", msg.channel_id)); } if !chan.get().is_usable() { @@ -2200,11 +2134,10 @@ impl ChannelManager { fn internal_update_fulfill_htlc(&self, their_node_id: &PublicKey, msg: &msgs::UpdateFulfillHTLC) -> Result<(), MsgHandleErrInternal> { let mut channel_lock = self.channel_state.lock().unwrap(); let htlc_source = { - let channel_state = channel_lock.borrow_parts(); + let channel_state = &mut *channel_lock; match channel_state.by_id.entry(msg.channel_id) { hash_map::Entry::Occupied(mut chan) => { if chan.get().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)); } try_chan_entry!(self, chan.get_mut().update_fulfill_htlc(&msg), channel_state, chan) @@ -2218,11 +2151,10 @@ impl ChannelManager { fn internal_update_fail_htlc(&self, their_node_id: &PublicKey, msg: &msgs::UpdateFailHTLC) -> Result<(), MsgHandleErrInternal> { let mut channel_lock = self.channel_state.lock().unwrap(); - let channel_state = channel_lock.borrow_parts(); + let channel_state = &mut *channel_lock; match channel_state.by_id.entry(msg.channel_id) { hash_map::Entry::Occupied(mut chan) => { if chan.get().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)); } try_chan_entry!(self, chan.get_mut().update_fail_htlc(&msg, HTLCFailReason::LightningError { err: msg.reason.clone() }), channel_state, chan); @@ -2234,11 +2166,10 @@ impl ChannelManager { fn internal_update_fail_malformed_htlc(&self, their_node_id: &PublicKey, msg: &msgs::UpdateFailMalformedHTLC) -> Result<(), MsgHandleErrInternal> { let mut channel_lock = self.channel_state.lock().unwrap(); - let channel_state = channel_lock.borrow_parts(); + let channel_state = &mut *channel_lock; match channel_state.by_id.entry(msg.channel_id) { hash_map::Entry::Occupied(mut chan) => { if chan.get().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)); } if (msg.failure_code & 0x8000) == 0 { @@ -2253,11 +2184,10 @@ impl ChannelManager { 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(); + let channel_state = &mut *channel_state_lock; match channel_state.by_id.entry(msg.channel_id) { hash_map::Entry::Occupied(mut chan) => { if chan.get().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) = @@ -2330,11 +2260,10 @@ impl ChannelManager { 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(); + let channel_state = &mut *channel_state_lock; match channel_state.by_id.entry(msg.channel_id) { hash_map::Entry::Occupied(mut chan) => { if chan.get().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 was_frozen_for_monitor = chan.get().is_awaiting_monitor_update(); @@ -2375,11 +2304,10 @@ impl ChannelManager { fn internal_update_fee(&self, their_node_id: &PublicKey, msg: &msgs::UpdateFee) -> Result<(), MsgHandleErrInternal> { let mut channel_lock = self.channel_state.lock().unwrap(); - let channel_state = channel_lock.borrow_parts(); + let channel_state = &mut *channel_lock; match channel_state.by_id.entry(msg.channel_id) { hash_map::Entry::Occupied(mut chan) => { if chan.get().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)); } try_chan_entry!(self, chan.get_mut().update_fee(&*self.fee_estimator, &msg), channel_state, chan); @@ -2391,7 +2319,7 @@ impl ChannelManager { fn internal_announcement_signatures(&self, their_node_id: &PublicKey, msg: &msgs::AnnouncementSignatures) -> Result<(), MsgHandleErrInternal> { let mut channel_state_lock = self.channel_state.lock().unwrap(); - let channel_state = channel_state_lock.borrow_parts(); + let channel_state = &mut *channel_state_lock; match channel_state.by_id.entry(msg.channel_id) { hash_map::Entry::Occupied(mut chan) => { @@ -2433,7 +2361,7 @@ impl ChannelManager { 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(); + let channel_state = &mut *channel_state_lock; match channel_state.by_id.entry(msg.channel_id) { hash_map::Entry::Occupied(mut chan) => { @@ -2508,10 +2436,10 @@ impl ChannelManager { #[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 their_node_id; let err: Result<(), _> = loop { - let mut channel_state_lock = self.channel_state.lock().unwrap(); - let channel_state = channel_state_lock.borrow_parts(); + let channel_state = &mut *channel_state_lock; match channel_state.by_id.entry(channel_id) { hash_map::Entry::Vacant(_) => return Err(APIError::APIMisuseError{err: "Failed to find corresponding channel"}), @@ -2549,20 +2477,9 @@ impl ChannelManager { return Ok(()) }; - match handle_error!(self, err) { + match handle_error!(self, err, their_node_id, channel_state_lock) { Ok(_) => unreachable!(), - Err(e) => { - if let msgs::ErrorAction::IgnoreError = e.action { - } else { - log_error!(self, "Got bad keys: {}!", e.err); - let mut channel_state = self.channel_state.lock().unwrap(); - channel_state.pending_msg_events.push(events::MessageSendEvent::HandleError { - node_id: their_node_id, - action: e.action, - }); - } - Err(APIError::APIMisuseError { err: e.err }) - }, + Err(e) => { Err(APIError::APIMisuseError { err: e.err })} } } } @@ -2625,9 +2542,9 @@ impl ChainListener for ChannelManager { let mut failed_channels = Vec::new(); { let mut channel_lock = self.channel_state.lock().unwrap(); - let channel_state = channel_lock.borrow_parts(); - let short_to_id = channel_state.short_to_id; - let pending_msg_events = channel_state.pending_msg_events; + let channel_state = &mut *channel_lock; + let short_to_id = &mut channel_state.short_to_id; + let pending_msg_events = &mut channel_state.pending_msg_events; 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 { @@ -2703,9 +2620,9 @@ impl ChainListener for ChannelManager { let mut failed_channels = Vec::new(); { let mut channel_lock = self.channel_state.lock().unwrap(); - let channel_state = channel_lock.borrow_parts(); - let short_to_id = channel_state.short_to_id; - let pending_msg_events = channel_state.pending_msg_events; + let channel_state = &mut *channel_lock; + let short_to_id = &mut channel_state.short_to_id; + let pending_msg_events = &mut channel_state.pending_msg_events; channel_state.by_id.retain(|_, v| { if v.block_disconnected(header) { if let Some(short_id) = v.get_short_channel_id() { @@ -2732,96 +2649,160 @@ 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, their_local_features: LocalFeatures, msg: &msgs::OpenChannel) -> Result<(), LightningError> { + fn handle_open_channel(&self, their_node_id: &PublicKey, their_features: InitFeatures, msg: &msgs::OpenChannel) { let _ = self.total_consistency_lock.read().unwrap(); - handle_error!(self, self.internal_open_channel(their_node_id, their_local_features, msg)) + let res = self.internal_open_channel(their_node_id, their_features, msg); + if res.is_err() { + let mut channel_state_lock = self.channel_state.lock().unwrap(); + let _ = handle_error!(self, res, *their_node_id, channel_state_lock); + } } - fn handle_accept_channel(&self, their_node_id: &PublicKey, their_local_features: LocalFeatures, msg: &msgs::AcceptChannel) -> Result<(), LightningError> { + fn handle_accept_channel(&self, their_node_id: &PublicKey, their_features: InitFeatures, msg: &msgs::AcceptChannel) { let _ = self.total_consistency_lock.read().unwrap(); - handle_error!(self, self.internal_accept_channel(their_node_id, their_local_features, msg)) + let res = self.internal_accept_channel(their_node_id, their_features, msg); + if res.is_err() { + let mut channel_state_lock = self.channel_state.lock().unwrap(); + let _ = handle_error!(self, res, *their_node_id, channel_state_lock); + } } - fn handle_funding_created(&self, their_node_id: &PublicKey, msg: &msgs::FundingCreated) -> Result<(), LightningError> { + fn handle_funding_created(&self, their_node_id: &PublicKey, msg: &msgs::FundingCreated) { let _ = self.total_consistency_lock.read().unwrap(); - handle_error!(self, self.internal_funding_created(their_node_id, msg)) + let res = self.internal_funding_created(their_node_id, msg); + if res.is_err() { + let mut channel_state_lock = self.channel_state.lock().unwrap(); + let _ = handle_error!(self, res, *their_node_id, channel_state_lock); + } } - fn handle_funding_signed(&self, their_node_id: &PublicKey, msg: &msgs::FundingSigned) -> Result<(), LightningError> { + fn handle_funding_signed(&self, their_node_id: &PublicKey, msg: &msgs::FundingSigned) { let _ = self.total_consistency_lock.read().unwrap(); - handle_error!(self, self.internal_funding_signed(their_node_id, msg)) + let res = self.internal_funding_signed(their_node_id, msg); + if res.is_err() { + let mut channel_state_lock = self.channel_state.lock().unwrap(); + let _ = handle_error!(self, res, *their_node_id, channel_state_lock); + } } - fn handle_funding_locked(&self, their_node_id: &PublicKey, msg: &msgs::FundingLocked) -> Result<(), LightningError> { + fn handle_funding_locked(&self, their_node_id: &PublicKey, msg: &msgs::FundingLocked) { let _ = self.total_consistency_lock.read().unwrap(); - handle_error!(self, self.internal_funding_locked(their_node_id, msg)) + let res = self.internal_funding_locked(their_node_id, msg); + if res.is_err() { + let mut channel_state_lock = self.channel_state.lock().unwrap(); + let _ = handle_error!(self, res, *their_node_id, channel_state_lock); + } } - fn handle_shutdown(&self, their_node_id: &PublicKey, msg: &msgs::Shutdown) -> Result<(), LightningError> { + fn handle_shutdown(&self, their_node_id: &PublicKey, msg: &msgs::Shutdown) { let _ = self.total_consistency_lock.read().unwrap(); - handle_error!(self, self.internal_shutdown(their_node_id, msg)) + let res = self.internal_shutdown(their_node_id, msg); + if res.is_err() { + let mut channel_state_lock = self.channel_state.lock().unwrap(); + let _ = handle_error!(self, res, *their_node_id, channel_state_lock); + } } - fn handle_closing_signed(&self, their_node_id: &PublicKey, msg: &msgs::ClosingSigned) -> Result<(), LightningError> { + fn handle_closing_signed(&self, their_node_id: &PublicKey, msg: &msgs::ClosingSigned) { let _ = self.total_consistency_lock.read().unwrap(); - handle_error!(self, self.internal_closing_signed(their_node_id, msg)) + let res = self.internal_closing_signed(their_node_id, msg); + if res.is_err() { + let mut channel_state_lock = self.channel_state.lock().unwrap(); + let _ = handle_error!(self, res, *their_node_id, channel_state_lock); + } } - fn handle_update_add_htlc(&self, their_node_id: &PublicKey, msg: &msgs::UpdateAddHTLC) -> Result<(), LightningError> { + fn handle_update_add_htlc(&self, their_node_id: &PublicKey, msg: &msgs::UpdateAddHTLC) { let _ = self.total_consistency_lock.read().unwrap(); - handle_error!(self, self.internal_update_add_htlc(their_node_id, msg)) + let res = self.internal_update_add_htlc(their_node_id, msg); + if res.is_err() { + let mut channel_state_lock = self.channel_state.lock().unwrap(); + let _ = handle_error!(self, res, *their_node_id, channel_state_lock); + } } - fn handle_update_fulfill_htlc(&self, their_node_id: &PublicKey, msg: &msgs::UpdateFulfillHTLC) -> Result<(), LightningError> { + fn handle_update_fulfill_htlc(&self, their_node_id: &PublicKey, msg: &msgs::UpdateFulfillHTLC) { let _ = self.total_consistency_lock.read().unwrap(); - handle_error!(self, self.internal_update_fulfill_htlc(their_node_id, msg)) + let res = self.internal_update_fulfill_htlc(their_node_id, msg); + if res.is_err() { + let mut channel_state_lock = self.channel_state.lock().unwrap(); + let _ = handle_error!(self, res, *their_node_id, channel_state_lock); + } } - fn handle_update_fail_htlc(&self, their_node_id: &PublicKey, msg: &msgs::UpdateFailHTLC) -> Result<(), LightningError> { + fn handle_update_fail_htlc(&self, their_node_id: &PublicKey, msg: &msgs::UpdateFailHTLC) { let _ = self.total_consistency_lock.read().unwrap(); - handle_error!(self, self.internal_update_fail_htlc(their_node_id, msg)) + let res = self.internal_update_fail_htlc(their_node_id, msg); + if res.is_err() { + let mut channel_state_lock = self.channel_state.lock().unwrap(); + let _ = handle_error!(self, res, *their_node_id, channel_state_lock); + } } - fn handle_update_fail_malformed_htlc(&self, their_node_id: &PublicKey, msg: &msgs::UpdateFailMalformedHTLC) -> Result<(), LightningError> { + fn handle_update_fail_malformed_htlc(&self, their_node_id: &PublicKey, msg: &msgs::UpdateFailMalformedHTLC) { let _ = self.total_consistency_lock.read().unwrap(); - handle_error!(self, self.internal_update_fail_malformed_htlc(their_node_id, msg)) + let res = self.internal_update_fail_malformed_htlc(their_node_id, msg); + if res.is_err() { + let mut channel_state_lock = self.channel_state.lock().unwrap(); + let _ = handle_error!(self, res, *their_node_id, channel_state_lock); + } } - fn handle_commitment_signed(&self, their_node_id: &PublicKey, msg: &msgs::CommitmentSigned) -> Result<(), LightningError> { + fn handle_commitment_signed(&self, their_node_id: &PublicKey, msg: &msgs::CommitmentSigned) { let _ = self.total_consistency_lock.read().unwrap(); - handle_error!(self, self.internal_commitment_signed(their_node_id, msg)) + let res = self.internal_commitment_signed(their_node_id, msg); + if res.is_err() { + let mut channel_state_lock = self.channel_state.lock().unwrap(); + let _ = handle_error!(self, res, *their_node_id, channel_state_lock); + } } - fn handle_revoke_and_ack(&self, their_node_id: &PublicKey, msg: &msgs::RevokeAndACK) -> Result<(), LightningError> { + fn handle_revoke_and_ack(&self, their_node_id: &PublicKey, msg: &msgs::RevokeAndACK) { let _ = self.total_consistency_lock.read().unwrap(); - handle_error!(self, self.internal_revoke_and_ack(their_node_id, msg)) + let res = self.internal_revoke_and_ack(their_node_id, msg); + if res.is_err() { + let mut channel_state_lock = self.channel_state.lock().unwrap(); + let _ = handle_error!(self, res, *their_node_id, channel_state_lock); + } } - fn handle_update_fee(&self, their_node_id: &PublicKey, msg: &msgs::UpdateFee) -> Result<(), LightningError> { + fn handle_update_fee(&self, their_node_id: &PublicKey, msg: &msgs::UpdateFee) { let _ = self.total_consistency_lock.read().unwrap(); - handle_error!(self, self.internal_update_fee(their_node_id, msg)) + let res = self.internal_update_fee(their_node_id, msg); + if res.is_err() { + let mut channel_state_lock = self.channel_state.lock().unwrap(); + let _ = handle_error!(self, res, *their_node_id, channel_state_lock); + } } - fn handle_announcement_signatures(&self, their_node_id: &PublicKey, msg: &msgs::AnnouncementSignatures) -> Result<(), LightningError> { + fn handle_announcement_signatures(&self, their_node_id: &PublicKey, msg: &msgs::AnnouncementSignatures) { let _ = self.total_consistency_lock.read().unwrap(); - handle_error!(self, self.internal_announcement_signatures(their_node_id, msg)) + let res = self.internal_announcement_signatures(their_node_id, msg); + if res.is_err() { + let mut channel_state_lock = self.channel_state.lock().unwrap(); + let _ = handle_error!(self, res, *their_node_id, channel_state_lock); + } } - fn handle_channel_reestablish(&self, their_node_id: &PublicKey, msg: &msgs::ChannelReestablish) -> Result<(), LightningError> { + fn handle_channel_reestablish(&self, their_node_id: &PublicKey, msg: &msgs::ChannelReestablish) { let _ = self.total_consistency_lock.read().unwrap(); - handle_error!(self, self.internal_channel_reestablish(their_node_id, msg)) + let res = self.internal_channel_reestablish(their_node_id, msg); + if res.is_err() { + let mut channel_state_lock = self.channel_state.lock().unwrap(); + let _ = handle_error!(self, res, *their_node_id, channel_state_lock); + } } 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(); + let mut no_channels_remain = true; { 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; - let pending_msg_events = channel_state.pending_msg_events; + let channel_state = &mut *channel_state_lock; + let short_to_id = &mut channel_state.short_to_id; + let pending_msg_events = &mut channel_state.pending_msg_events; if no_connection_possible { log_debug!(self, "Failing all channels with {} due to no_connection_possible", log_pubkey!(their_node_id)); channel_state.by_id.retain(|_, chan| { @@ -2855,6 +2836,8 @@ impl ChannelMessageHandler for ChannelManager ChannelMessageHandler for ChannelManager ChannelMessageHandler for ChannelManager { + e.insert(Mutex::new(PeerState { + latest_features: init_msg.features.clone(), + })); + }, + hash_map::Entry::Occupied(e) => { + e.get().lock().unwrap().latest_features = init_msg.features.clone(); + }, + } + } + 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; + let channel_state = &mut *channel_state_lock; + let pending_msg_events = &mut 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() { @@ -3160,6 +3162,14 @@ impl Writeable for ChannelManager { /// /// In such cases the latest local transactions will be sent to the tx_broadcaster included in /// this struct. - pub channel_monitors: &'a HashMap, + pub channel_monitors: &'a mut HashMap, } impl<'a, R : ::std::io::Read, ChanSigner: ChannelKeys + Readable> ReadableArgs> for (Sha256dHash, ChannelManager) { @@ -3245,7 +3255,7 @@ impl<'a, R : ::std::io::Read, ChanSigner: ChannelKeys + Readable> ReadableArg 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 let Some(ref mut monitor) = args.channel_monitors.get_mut(&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() { @@ -3263,7 +3273,7 @@ impl<'a, R : ::std::io::Read, ChanSigner: ChannelKeys + Readable> ReadableArg } } - for (ref funding_txo, ref monitor) in args.channel_monitors.iter() { + for (ref funding_txo, ref mut monitor) in args.channel_monitors.iter_mut() { if !funding_txo_set.contains(funding_txo) { closed_channels.push((monitor.get_latest_local_commitment_txn(), Vec::new())); } @@ -3293,6 +3303,16 @@ impl<'a, R : ::std::io::Read, ChanSigner: ChannelKeys + Readable> ReadableArg claimable_htlcs.insert(payment_hash, previous_hops); } + let peer_count: u64 = Readable::read(reader)?; + let mut per_peer_state = HashMap::with_capacity(cmp::min(peer_count as usize, 128)); + for _ in 0..peer_count { + let peer_pubkey = Readable::read(reader)?; + let peer_state = PeerState { + latest_features: Readable::read(reader)?, + }; + per_peer_state.insert(peer_pubkey, Mutex::new(peer_state)); + } + let channel_manager = ChannelManager { genesis_hash, fee_estimator: args.fee_estimator, @@ -3312,6 +3332,8 @@ impl<'a, R : ::std::io::Read, ChanSigner: ChannelKeys + Readable> ReadableArg }), our_network_key: args.keys_manager.get_node_secret(), + per_peer_state: RwLock::new(per_peer_state), + pending_events: Mutex::new(Vec::new()), total_consistency_lock: RwLock::new(()), keys_manager: args.keys_manager,