+ for failure in pending_failures.drain(..) {
+ self.fail_htlc_backwards_internal(self.channel_state.lock().unwrap(), failure.0, &failure.1, failure.2);
+ }
+
+ let mut forward_event = None;
+ 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();
+ }
+ for (forward_info, prev_htlc_id) in pending_forwards.drain(..) {
+ match channel_state.forward_htlcs.entry(forward_info.short_channel_id) {
+ hash_map::Entry::Occupied(mut entry) => {
+ entry.get_mut().push(HTLCForwardInfo { prev_short_channel_id: short_channel_id, prev_htlc_id, forward_info });
+ },
+ hash_map::Entry::Vacant(entry) => {
+ entry.insert(vec!(HTLCForwardInfo { prev_short_channel_id: short_channel_id, prev_htlc_id, forward_info }));
+ }
+ }
+ }
+ }
+ match forward_event {
+ Some(time) => {
+ let mut pending_events = self.pending_events.lock().unwrap();
+ pending_events.push(events::Event::PendingHTLCsForwardable {
+ time_forwardable: time
+ });
+ }
+ None => {},
+ }
+
+ Ok(res)
+ }
+
+ fn internal_update_fee(&self, their_node_id: &PublicKey, msg: &msgs::UpdateFee) -> Result<(), MsgHandleErrInternal> {
+ let mut channel_state = self.channel_state.lock().unwrap();
+ match channel_state.by_id.get_mut(&msg.channel_id) {
+ Some(chan) => {
+ if chan.get_their_node_id() != *their_node_id {
+ //TODO: here and below MsgHandleErrInternal, #153 case
+ return Err(MsgHandleErrInternal::send_err_msg_no_close("Got a message for a channel from the wrong node!", msg.channel_id));
+ }
+ chan.update_fee(&*self.fee_estimator, &msg).map_err(|e| MsgHandleErrInternal::from_maybe_close(e))
+ },
+ None => return Err(MsgHandleErrInternal::send_err_msg_no_close("Failed to find corresponding channel", msg.channel_id))
+ }
+ }
+
+ fn internal_announcement_signatures(&self, their_node_id: &PublicKey, msg: &msgs::AnnouncementSignatures) -> Result<(), MsgHandleErrInternal> {
+ let (chan_announcement, chan_update) = {
+ let mut channel_state = self.channel_state.lock().unwrap();
+ match channel_state.by_id.get_mut(&msg.channel_id) {
+ Some(chan) => {
+ if chan.get_their_node_id() != *their_node_id {
+ return Err(MsgHandleErrInternal::send_err_msg_no_close("Got a message for a channel from the wrong node!", msg.channel_id));
+ }
+ if !chan.is_usable() {
+ return Err(MsgHandleErrInternal::from_no_close(HandleError{err: "Got an announcement_signatures before we were ready for it", action: Some(msgs::ErrorAction::IgnoreError)}));
+ }
+
+ let our_node_id = self.get_our_node_id();
+ let (announcement, our_bitcoin_sig) = chan.get_channel_announcement(our_node_id.clone(), self.genesis_hash.clone())
+ .map_err(|e| MsgHandleErrInternal::from_maybe_close(e))?;
+
+ let were_node_one = announcement.node_id_1 == our_node_id;
+ let msghash = Message::from_slice(&Sha256dHash::from_data(&announcement.encode()[..])[..]).unwrap();
+ let bad_sig_action = MsgHandleErrInternal::send_err_msg_close_chan("Bad announcement_signatures node_signature", msg.channel_id);
+ secp_call!(self.secp_ctx.verify(&msghash, &msg.node_signature, if were_node_one { &announcement.node_id_2 } else { &announcement.node_id_1 }), bad_sig_action);
+ secp_call!(self.secp_ctx.verify(&msghash, &msg.bitcoin_signature, if were_node_one { &announcement.bitcoin_key_2 } else { &announcement.bitcoin_key_1 }), bad_sig_action);
+
+ let our_node_sig = self.secp_ctx.sign(&msghash, &self.our_network_key);
+
+ (msgs::ChannelAnnouncement {
+ node_signature_1: if were_node_one { our_node_sig } else { msg.node_signature },
+ node_signature_2: if were_node_one { msg.node_signature } else { our_node_sig },
+ bitcoin_signature_1: if were_node_one { our_bitcoin_sig } else { msg.bitcoin_signature },
+ bitcoin_signature_2: if were_node_one { msg.bitcoin_signature } else { our_bitcoin_sig },
+ contents: announcement,
+ }, self.get_channel_update(chan).unwrap()) // can only fail if we're not in a ready state
+ },
+ None => return Err(MsgHandleErrInternal::send_err_msg_no_close("Failed to find corresponding channel", msg.channel_id))
+ }
+ };
+ let mut pending_events = self.pending_events.lock().unwrap();
+ pending_events.push(events::Event::BroadcastChannelAnnouncement { msg: chan_announcement, update_msg: chan_update });
+ Ok(())
+ }
+
+
+}
+
+impl events::EventsProvider for ChannelManager {
+ fn get_and_clear_pending_events(&self) -> Vec<events::Event> {
+ let mut pending_events = self.pending_events.lock().unwrap();
+ let mut ret = Vec::new();
+ mem::swap(&mut ret, &mut *pending_events);
+ ret
+ }
+}
+
+impl ChainListener for ChannelManager {
+ fn block_connected(&self, header: &BlockHeader, height: u32, txn_matched: &[&Transaction], indexes_of_txn_matched: &[u32]) {
+ let mut new_events = Vec::new();
+ 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;
+ channel_state.by_id.retain(|_, channel| {
+ let chan_res = channel.block_connected(header, height, txn_matched, indexes_of_txn_matched);
+ if let Ok(Some(funding_locked)) = chan_res {
+ let announcement_sigs = self.get_announcement_sigs(channel);
+ new_events.push(events::Event::SendFundingLocked {
+ node_id: channel.get_their_node_id(),
+ msg: funding_locked,
+ announcement_sigs: announcement_sigs
+ });
+ short_to_id.insert(channel.get_short_channel_id().unwrap(), channel.channel_id());
+ } else if let Err(e) = chan_res {
+ new_events.push(events::Event::HandleError {
+ node_id: channel.get_their_node_id(),
+ action: e.action,
+ });
+ if channel.is_shutdown() {
+ return false;
+ }
+ }
+ if let Some(funding_txo) = channel.get_funding_txo() {
+ for tx in txn_matched {
+ for inp in tx.input.iter() {
+ if inp.previous_output == funding_txo.into_bitcoin_outpoint() {
+ if let Some(short_id) = channel.get_short_channel_id() {
+ short_to_id.remove(&short_id);
+ }
+ // It looks like our counterparty went on-chain. We go ahead and
+ // broadcast our latest local state as well here, just in case its
+ // some kind of SPV attack, though we expect these to be dropped.
+ failed_channels.push(channel.force_shutdown());
+ if let Ok(update) = self.get_channel_update(&channel) {
+ new_events.push(events::Event::BroadcastChannelUpdate {
+ msg: update
+ });
+ }
+ return false;
+ }
+ }
+ }
+ }
+ if channel.is_funding_initiated() && channel.channel_monitor().would_broadcast_at_height(height) {
+ if let Some(short_id) = channel.get_short_channel_id() {
+ short_to_id.remove(&short_id);
+ }
+ failed_channels.push(channel.force_shutdown());
+ // If would_broadcast_at_height() is true, the channel_monitor will broadcast
+ // the latest local tx for us, so we should skip that here (it doesn't really
+ // hurt anything, but does make tests a bit simpler).
+ failed_channels.last_mut().unwrap().0 = Vec::new();
+ if let Ok(update) = self.get_channel_update(&channel) {
+ new_events.push(events::Event::BroadcastChannelUpdate {
+ msg: update
+ });
+ }
+ return false;
+ }
+ true
+ });
+ }
+ for failure in failed_channels.drain(..) {
+ self.finish_force_close_channel(failure);
+ }
+ let mut pending_events = self.pending_events.lock().unwrap();
+ for funding_locked in new_events.drain(..) {
+ pending_events.push(funding_locked);
+ }
+ self.latest_block_height.store(height as usize, Ordering::Release);
+ }
+
+ /// We force-close the channel without letting our counterparty participate in the shutdown
+ fn block_disconnected(&self, header: &BlockHeader) {
+ let mut new_events = Vec::new();
+ 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;
+ channel_state.by_id.retain(|_, v| {
+ if v.block_disconnected(header) {
+ if let Some(short_id) = v.get_short_channel_id() {
+ short_to_id.remove(&short_id);
+ }
+ failed_channels.push(v.force_shutdown());
+ if let Ok(update) = self.get_channel_update(&v) {
+ new_events.push(events::Event::BroadcastChannelUpdate {
+ msg: update
+ });
+ }
+ false
+ } else {
+ true
+ }
+ });
+ }
+ for failure in failed_channels.drain(..) {
+ self.finish_force_close_channel(failure);
+ }
+ if !new_events.is_empty() {
+ let mut pending_events = self.pending_events.lock().unwrap();
+ for funding_locked in new_events.drain(..) {
+ pending_events.push(funding_locked);
+ }
+ }
+ self.latest_block_height.fetch_sub(1, Ordering::AcqRel);
+ }
+}
+
+macro_rules! handle_error {
+ ($self: ident, $internal: expr, $their_node_id: expr) => {
+ match $internal {
+ Ok(msg) => Ok(msg),
+ Err(MsgHandleErrInternal { err, needs_channel_force_close }) => {
+ if needs_channel_force_close {
+ match &err.action {
+ &Some(msgs::ErrorAction::DisconnectPeer { msg: Some(ref msg) }) => {
+ if msg.channel_id == [0; 32] {
+ $self.peer_disconnected(&$their_node_id, true);
+ } else {
+ $self.force_close_channel(&msg.channel_id);
+ }
+ },
+ &Some(msgs::ErrorAction::DisconnectPeer { msg: None }) => {},
+ &Some(msgs::ErrorAction::IgnoreError) => {},
+ &Some(msgs::ErrorAction::SendErrorMessage { ref msg }) => {
+ if msg.channel_id == [0; 32] {
+ $self.peer_disconnected(&$their_node_id, true);
+ } else {
+ $self.force_close_channel(&msg.channel_id);
+ }
+ },
+ &None => {},
+ }
+ }
+ Err(err)
+ },
+ }
+ }
+}
+
+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<msgs::AcceptChannel, HandleError> {
+ handle_error!(self, self.internal_open_channel(their_node_id, msg), their_node_id)
+ }
+
+ fn handle_accept_channel(&self, their_node_id: &PublicKey, msg: &msgs::AcceptChannel) -> Result<(), HandleError> {
+ handle_error!(self, self.internal_accept_channel(their_node_id, msg), their_node_id)
+ }
+
+ fn handle_funding_created(&self, their_node_id: &PublicKey, msg: &msgs::FundingCreated) -> Result<msgs::FundingSigned, HandleError> {
+ handle_error!(self, self.internal_funding_created(their_node_id, msg), their_node_id)
+ }
+
+ fn handle_funding_signed(&self, their_node_id: &PublicKey, msg: &msgs::FundingSigned) -> Result<(), HandleError> {
+ handle_error!(self, self.internal_funding_signed(their_node_id, msg), their_node_id)
+ }
+
+ fn handle_funding_locked(&self, their_node_id: &PublicKey, msg: &msgs::FundingLocked) -> Result<Option<msgs::AnnouncementSignatures>, HandleError> {
+ handle_error!(self, self.internal_funding_locked(their_node_id, msg), their_node_id)
+ }
+
+ fn handle_shutdown(&self, their_node_id: &PublicKey, msg: &msgs::Shutdown) -> Result<(Option<msgs::Shutdown>, Option<msgs::ClosingSigned>), HandleError> {
+ handle_error!(self, self.internal_shutdown(their_node_id, msg), their_node_id)
+ }
+
+ fn handle_closing_signed(&self, their_node_id: &PublicKey, msg: &msgs::ClosingSigned) -> Result<Option<msgs::ClosingSigned>, HandleError> {
+ handle_error!(self, self.internal_closing_signed(their_node_id, msg), their_node_id)
+ }
+
+ fn handle_update_add_htlc(&self, their_node_id: &PublicKey, msg: &msgs::UpdateAddHTLC) -> Result<(), msgs::HandleError> {
+ handle_error!(self, self.internal_update_add_htlc(their_node_id, msg), their_node_id)
+ }
+
+ fn handle_update_fulfill_htlc(&self, their_node_id: &PublicKey, msg: &msgs::UpdateFulfillHTLC) -> Result<(), HandleError> {
+ handle_error!(self, self.internal_update_fulfill_htlc(their_node_id, msg), their_node_id)
+ }