Refactor/dont re-enter block_conencted on duplicate watch calls
[rust-lightning] / src / ln / peer_handler.rs
index 273a36dafd4369f7092c42b5ccc26a523823c32a..3122e572e57fafcbd90e1c08c6c10b146d152c29 100644 (file)
@@ -5,7 +5,7 @@ use ln::msgs::{MsgEncodable,MsgDecodable};
 use ln::peer_channel_encryptor::{PeerChannelEncryptor,NextNoiseStep};
 use util::byte_utils;
 use util::events::{EventsProvider,Event};
-use util::logger::{Logger, Record};
+use util::logger::Logger;
 
 use std::collections::{HashMap,LinkedList};
 use std::sync::{Arc, Mutex};
@@ -289,7 +289,10 @@ impl<Descriptor: SocketDescriptor> PeerManager<Descriptor> {
 
                                                        macro_rules! encode_and_send_msg {
                                                                ($msg: expr, $msg_code: expr) => {
-                                                                       peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!($msg, $msg_code)[..]));
+                                                                       {
+                                                                               log_trace!(self, "Encoding and sending message of type {} to {}", $msg_code, log_pubkey!(peer.their_node_id.unwrap()));
+                                                                               peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!($msg, $msg_code)[..]));
+                                                                       }
                                                                }
                                                        }
 
@@ -300,22 +303,23 @@ impl<Descriptor: SocketDescriptor> PeerManager<Descriptor> {
                                                                                Err(e) => {
                                                                                        if let Some(action) = e.action {
                                                                                                match action {
-                                                                                                       msgs::ErrorAction::UpdateFailHTLC { msg } => {
-                                                                                                               encode_and_send_msg!(msg, 131);
-                                                                                                               continue;
-                                                                                                       },
                                                                                                        msgs::ErrorAction::DisconnectPeer { msg: _ } => {
+                                                                                                               //TODO: Try to push msg
+                                                                                                               log_trace!(self, "Got Err handling message, disconnecting peer because {}", e.err);
                                                                                                                return Err(PeerHandleError{ no_connection_possible: false });
                                                                                                        },
                                                                                                        msgs::ErrorAction::IgnoreError => {
+                                                                                                               log_trace!(self, "Got Err handling message, ignoring because {}", e.err);
                                                                                                                continue;
                                                                                                        },
                                                                                                        msgs::ErrorAction::SendErrorMessage { msg } => {
+                                                                                                               log_trace!(self, "Got Err handling message, sending Error message because {}", e.err);
                                                                                                                encode_and_send_msg!(msg, 17);
                                                                                                                continue;
                                                                                                        },
                                                                                                }
                                                                                        } else {
+                                                                                               log_debug!(self, "Got Err handling message, action not yet filled in: {}", e.err);
                                                                                                return Err(PeerHandleError{ no_connection_possible: false });
                                                                                        }
                                                                                }
@@ -327,21 +331,23 @@ impl<Descriptor: SocketDescriptor> PeerManager<Descriptor> {
                                                                ($thing: expr) => {
                                                                        match $thing {
                                                                                Ok(x) => x,
-                                                                               Err(_e) => {
-                                                                                       //TODO: Handle e?
-                                                                                       return Err(PeerHandleError{ no_connection_possible: false });
-                                                                               }
-                                                                       };
-                                                               }
-                                                       }
-
-                                                       macro_rules! try_ignore_potential_decodeerror {
-                                                               ($thing: expr) => {
-                                                                       match $thing {
-                                                                               Ok(x) => x,
-                                                                               Err(_e) => {
-                                                                                       log_debug!(self, "Error decoding message, ignoring due to lnd spec incompatibility. See https://github.com/lightningnetwork/lnd/issues/1407");
-                                                                                       continue;
+                                                                               Err(e) => {
+                                                                                       match e {
+                                                                                               msgs::DecodeError::UnknownRealmByte => return Err(PeerHandleError{ no_connection_possible: false }),
+                                                                                               msgs::DecodeError::UnknownRequiredFeature => {
+                                                                                                       log_debug!(self, "Got a channel/node announcement with an known required feature flag, you may want to udpate!");
+                                                                                                       continue;
+                                                                                               },
+                                                                                               msgs::DecodeError::BadPublicKey => return Err(PeerHandleError{ no_connection_possible: false }),
+                                                                                               msgs::DecodeError::BadSignature => return Err(PeerHandleError{ no_connection_possible: false }),
+                                                                                               msgs::DecodeError::BadText => return Err(PeerHandleError{ no_connection_possible: false }),
+                                                                                               msgs::DecodeError::ShortRead => return Err(PeerHandleError{ no_connection_possible: false }),
+                                                                                               msgs::DecodeError::ExtraAddressesPerType => {
+                                                                                                       log_debug!(self, "Error decoding message, ignoring due to lnd spec incompatibility. See https://github.com/lightningnetwork/lnd/issues/1407");
+                                                                                                       continue;
+                                                                                               },
+                                                                                               msgs::DecodeError::BadLengthDescriptor => return Err(PeerHandleError{ no_connection_possible: false }),
+                                                                                       }
                                                                                }
                                                                        };
                                                                }
@@ -396,6 +402,7 @@ impl<Descriptor: SocketDescriptor> PeerManager<Descriptor> {
                                                                                peer.pending_read_is_header = true;
 
                                                                                let msg_type = byte_utils::slice_to_be16(&msg_data[0..2]);
+                                                                               log_trace!(self, "Received message of type {} from {}", msg_type, log_pubkey!(peer.their_node_id.unwrap()));
                                                                                if msg_type != 16 && peer.their_global_features.is_none() {
                                                                                        // Need an init message as first message
                                                                                        return Err(PeerHandleError{ no_connection_possible: false });
@@ -426,7 +433,24 @@ impl<Descriptor: SocketDescriptor> PeerManager<Descriptor> {
                                                                                                }
                                                                                        },
                                                                                        17 => {
-                                                                                               // Error msg
+                                                                                               let msg = try_potential_decodeerror!(msgs::ErrorMessage::decode(&msg_data[2..]));
+                                                                                               let mut data_is_printable = true;
+                                                                                               for b in msg.data.bytes() {
+                                                                                                       if b < 32 || b > 126 {
+                                                                                                               data_is_printable = false;
+                                                                                                               break;
+                                                                                                       }
+                                                                                               }
+
+                                                                                               if data_is_printable {
+                                                                                                       log_debug!(self, "Got Err message from {}: {}", log_pubkey!(peer.their_node_id.unwrap()), msg.data);
+                                                                                               } else {
+                                                                                                       log_debug!(self, "Got Err message from {} with non-ASCII error message", log_pubkey!(peer.their_node_id.unwrap()));
+                                                                                               }
+                                                                                               self.message_handler.chan_handler.handle_error(&peer.their_node_id.unwrap(), &msg);
+                                                                                               if msg.channel_id == [0; 32] {
+                                                                                                       return Err(PeerHandleError{ no_connection_possible: true });
+                                                                                               }
                                                                                        },
 
                                                                                        18 => {
@@ -554,12 +578,20 @@ impl<Descriptor: SocketDescriptor> PeerManager<Descriptor> {
                                                                                                }
                                                                                        },
                                                                                        257 => {
-                                                                                               let msg = try_ignore_potential_decodeerror!(msgs::NodeAnnouncement::decode(&msg_data[2..]));
-                                                                                               try_potential_handleerror!(self.message_handler.route_handler.handle_node_announcement(&msg));
+                                                                                               let msg = try_potential_decodeerror!(msgs::NodeAnnouncement::decode(&msg_data[2..]));
+                                                                                               let should_forward = try_potential_handleerror!(self.message_handler.route_handler.handle_node_announcement(&msg));
+
+                                                                                               if should_forward {
+                                                                                                       // TODO: forward msg along to all our other peers!
+                                                                                               }
                                                                                        },
                                                                                        258 => {
                                                                                                let msg = try_potential_decodeerror!(msgs::ChannelUpdate::decode(&msg_data[2..]));
-                                                                                               try_potential_handleerror!(self.message_handler.route_handler.handle_channel_update(&msg));
+                                                                                               let should_forward = try_potential_handleerror!(self.message_handler.route_handler.handle_channel_update(&msg));
+
+                                                                                               if should_forward {
+                                                                                                       // TODO: forward msg along to all our other peers!
+                                                                                               }
                                                                                        },
                                                                                        _ => {
                                                                                                if (msg_type & 1) == 0 {
@@ -616,6 +648,10 @@ impl<Descriptor: SocketDescriptor> PeerManager<Descriptor> {
                                                        };
                                                        match peers.peers.get_mut(&descriptor) {
                                                                Some(peer) => {
+                                                                       if peer.their_global_features.is_none() {
+                                                                               $handle_no_such_peer;
+                                                                               continue;
+                                                                       }
                                                                        (descriptor, peer)
                                                                },
                                                                None => panic!("Inconsistent peers set state!"),
@@ -632,6 +668,9 @@ impl<Descriptor: SocketDescriptor> PeerManager<Descriptor> {
                                        Event::PendingHTLCsForwardable {..} => { /* Hand upstream */ },
 
                                        Event::SendOpenChannel { ref node_id, ref msg } => {
+                                               log_trace!(self, "Handling SendOpenChannel event in peer_handler for node {} for channel {}",
+                                                               log_pubkey!(node_id),
+                                                               log_bytes!(msg.temporary_channel_id));
                                                let (mut descriptor, peer) = get_peer_for_forwarding!(node_id, {
                                                                //TODO: Drop the pending channel? (or just let it timeout, but that sucks)
                                                        });
@@ -640,6 +679,10 @@ impl<Descriptor: SocketDescriptor> PeerManager<Descriptor> {
                                                continue;
                                        },
                                        Event::SendFundingCreated { ref node_id, ref msg } => {
+                                               log_trace!(self, "Handling SendFundingCreated event in peer_handler for node {} for channel {} (which becomes {})",
+                                                               log_pubkey!(node_id),
+                                                               log_bytes!(msg.temporary_channel_id),
+                                                               log_funding_channel_id!(msg.funding_txid, msg.funding_output_index));
                                                let (mut descriptor, peer) = get_peer_for_forwarding!(node_id, {
                                                                //TODO: generate a DiscardFunding event indicating to the wallet that
                                                                //they should just throw away this funding transaction
@@ -649,6 +692,10 @@ impl<Descriptor: SocketDescriptor> PeerManager<Descriptor> {
                                                continue;
                                        },
                                        Event::SendFundingLocked { ref node_id, ref msg, ref announcement_sigs } => {
+                                               log_trace!(self, "Handling SendFundingLocked event in peer_handler for node {}{} for channel {}",
+                                                               log_pubkey!(node_id),
+                                                               if announcement_sigs.is_some() { " with announcement sigs" } else { "" },
+                                                               log_bytes!(msg.channel_id));
                                                let (mut descriptor, peer) = get_peer_for_forwarding!(node_id, {
                                                                //TODO: Do whatever we're gonna do for handling dropped messages
                                                        });
@@ -660,36 +707,36 @@ impl<Descriptor: SocketDescriptor> PeerManager<Descriptor> {
                                                Self::do_attempt_write_data(&mut descriptor, peer);
                                                continue;
                                        },
-                                       Event::SendHTLCs { ref node_id, ref msgs, ref commitment_msg } => {
+                                       Event::UpdateHTLCs { ref node_id, updates: msgs::CommitmentUpdate { ref update_add_htlcs, ref update_fulfill_htlcs, ref update_fail_htlcs, ref update_fail_malformed_htlcs, ref commitment_signed } } => {
+                                               log_trace!(self, "Handling UpdateHTLCs event in peer_handler for node {} with {} adds, {} fulfills, {} fails for channel {}",
+                                                               log_pubkey!(node_id),
+                                                               update_add_htlcs.len(),
+                                                               update_fulfill_htlcs.len(),
+                                                               update_fail_htlcs.len(),
+                                                               log_bytes!(commitment_signed.channel_id));
                                                let (mut descriptor, peer) = get_peer_for_forwarding!(node_id, {
                                                                //TODO: Do whatever we're gonna do for handling dropped messages
                                                        });
-                                               for msg in msgs {
+                                               for msg in update_add_htlcs {
                                                        peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg, 128)));
                                                }
-                                               peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(commitment_msg, 132)));
-                                               Self::do_attempt_write_data(&mut descriptor, peer);
-                                               continue;
-                                       },
-                                       Event::SendFulfillHTLC { ref node_id, ref msg, ref commitment_msg } => {
-                                               let (mut descriptor, peer) = get_peer_for_forwarding!(node_id, {
-                                                               //TODO: Do whatever we're gonna do for handling dropped messages
-                                                       });
-                                               peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg, 130)));
-                                               peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(commitment_msg, 132)));
-                                               Self::do_attempt_write_data(&mut descriptor, peer);
-                                               continue;
-                                       },
-                                       Event::SendFailHTLC { ref node_id, ref msg, ref commitment_msg } => {
-                                               let (mut descriptor, peer) = get_peer_for_forwarding!(node_id, {
-                                                               //TODO: Do whatever we're gonna do for handling dropped messages
-                                                       });
-                                               peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg, 131)));
-                                               peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(commitment_msg, 132)));
+                                               for msg in update_fulfill_htlcs {
+                                                       peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg, 130)));
+                                               }
+                                               for msg in update_fail_htlcs {
+                                                       peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg, 131)));
+                                               }
+                                               for msg in update_fail_malformed_htlcs {
+                                                       peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg, 135)));
+                                               }
+                                               peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(commitment_signed, 132)));
                                                Self::do_attempt_write_data(&mut descriptor, peer);
                                                continue;
                                        },
                                        Event::SendShutdown { ref node_id, ref msg } => {
+                                               log_trace!(self, "Handling Shutdown event in peer_handler for node {} for channel {}",
+                                                               log_pubkey!(node_id),
+                                                               log_bytes!(msg.channel_id));
                                                let (mut descriptor, peer) = get_peer_for_forwarding!(node_id, {
                                                                //TODO: Do whatever we're gonna do for handling dropped messages
                                                        });
@@ -698,12 +745,13 @@ impl<Descriptor: SocketDescriptor> PeerManager<Descriptor> {
                                                continue;
                                        },
                                        Event::BroadcastChannelAnnouncement { ref msg, ref update_msg } => {
+                                               log_trace!(self, "Handling BroadcastChannelAnnouncement event in peer_handler for short channel id {}", msg.contents.short_channel_id);
                                                if self.message_handler.route_handler.handle_channel_announcement(msg).is_ok() && self.message_handler.route_handler.handle_channel_update(update_msg).is_ok() {
                                                        let encoded_msg = encode_msg!(msg, 256);
                                                        let encoded_update_msg = encode_msg!(update_msg, 258);
 
                                                        for (ref descriptor, ref mut peer) in peers.peers.iter_mut() {
-                                                               if !peer.channel_encryptor.is_ready_for_encryption() {
+                                                               if !peer.channel_encryptor.is_ready_for_encryption() || peer.their_global_features.is_none() {
                                                                        continue
                                                                }
                                                                match peer.their_node_id {
@@ -722,11 +770,12 @@ impl<Descriptor: SocketDescriptor> PeerManager<Descriptor> {
                                                continue;
                                        },
                                        Event::BroadcastChannelUpdate { ref msg } => {
+                                               log_trace!(self, "Handling BroadcastChannelUpdate event in peer_handler for short channel id {}", msg.contents.short_channel_id);
                                                if self.message_handler.route_handler.handle_channel_update(msg).is_ok() {
                                                        let encoded_msg = encode_msg!(msg, 258);
 
                                                        for (ref descriptor, ref mut peer) in peers.peers.iter_mut() {
-                                                               if !peer.channel_encryptor.is_ready_for_encryption() {
+                                                               if !peer.channel_encryptor.is_ready_for_encryption() || peer.their_global_features.is_none() {
                                                                        continue
                                                                }
                                                                peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encoded_msg[..]));
@@ -738,22 +787,19 @@ impl<Descriptor: SocketDescriptor> PeerManager<Descriptor> {
                                        Event::HandleError { ref node_id, ref action } => {
                                                if let Some(ref action) = *action {
                                                        match *action {
-                                                               msgs::ErrorAction::UpdateFailHTLC { ref msg } => {
-                                                                       let (mut descriptor, peer) = get_peer_for_forwarding!(node_id, {
-                                                                               //TODO: Do whatever we're gonna do for handling dropped messages
-                                                                       });
-                                                                       peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg, 131)));
-                                                                       Self::do_attempt_write_data(&mut descriptor, peer);
-
-                                                               },
                                                                msgs::ErrorAction::DisconnectPeer { ref msg } => {
                                                                        if let Some(mut descriptor) = peers.node_id_to_descriptor.remove(node_id) {
                                                                                if let Some(mut peer) = peers.peers.remove(&descriptor) {
                                                                                        if let Some(ref msg) = *msg {
+                                                                                               log_trace!(self, "Handling DisconnectPeer HandleError event in peer_handler for node {} with message {}",
+                                                                                                               log_pubkey!(node_id),
+                                                                                                               msg.data);
                                                                                                peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg, 17)));
                                                                                                // This isn't guaranteed to work, but if there is enough free
                                                                                                // room in the send buffer, put the error message there...
                                                                                                Self::do_attempt_write_data(&mut descriptor, &mut peer);
+                                                                                       } else {
+                                                                                               log_trace!(self, "Handling DisconnectPeer HandleError event in peer_handler for node {} with no message", log_pubkey!(node_id));
                                                                                        }
                                                                                }
                                                                                descriptor.disconnect_socket();
@@ -764,6 +810,9 @@ impl<Descriptor: SocketDescriptor> PeerManager<Descriptor> {
                                                                        continue;
                                                                },
                                                                msgs::ErrorAction::SendErrorMessage { ref msg } => {
+                                                                       log_trace!(self, "Handling SendErrorMessage HandleError event in peer_handler for node {} with message {}",
+                                                                                       log_pubkey!(node_id),
+                                                                                       msg.data);
                                                                        let (mut descriptor, peer) = get_peer_for_forwarding!(node_id, {
                                                                                //TODO: Do whatever we're gonna do for handling dropped messages
                                                                        });
@@ -771,6 +820,8 @@ impl<Descriptor: SocketDescriptor> PeerManager<Descriptor> {
                                                                        Self::do_attempt_write_data(&mut descriptor, peer);
                                                                },
                                                        }
+                                               } else {
+                                                       log_error!(self, "Got no-action HandleError Event in peer_handler for node {}, no such events should ever be generated!", log_pubkey!(node_id));
                                                }
                                                continue;
                                        }
@@ -874,7 +925,7 @@ mod tests {
 
        fn establish_connection(peer_a: &PeerManager<FileDescriptor>, peer_b: &PeerManager<FileDescriptor>) {
                let secp_ctx = Secp256k1::new();
-               let their_id = PublicKey::from_secret_key(&secp_ctx, &peer_b.our_node_secret).unwrap();
+               let their_id = PublicKey::from_secret_key(&secp_ctx, &peer_b.our_node_secret);
                let fd = FileDescriptor { fd: 1};
                peer_a.new_inbound_connection(fd.clone()).unwrap();
                peer_a.peers.lock().unwrap().node_id_to_descriptor.insert(their_id, fd.clone());
@@ -889,7 +940,7 @@ mod tests {
                assert_eq!(peers[0].peers.lock().unwrap().peers.len(), 1);
 
                let secp_ctx = Secp256k1::new();
-               let their_id = PublicKey::from_secret_key(&secp_ctx, &peers[1].our_node_secret).unwrap();
+               let their_id = PublicKey::from_secret_key(&secp_ctx, &peers[1].our_node_secret);
 
                let chan_handler = test_utils::TestChannelMessageHandler::new();
                chan_handler.pending_events.lock().unwrap().push(events::Event::HandleError {