X-Git-Url: http://git.bitcoin.ninja/index.cgi?a=blobdiff_plain;f=src%2Fln%2Fpeer_handler.rs;h=94cdef9e41fe76f7fa568edba784961edaeb8825;hb=f1335546972b79082ab762b10a92c2c4e4f09ac6;hp=6c2b4d798c4a002ac52c359c2fe2d5d067eefafd;hpb=66fbc66da0c064e4ee271ad5a8e0a37410c80078;p=rust-lightning diff --git a/src/ln/peer_handler.rs b/src/ln/peer_handler.rs index 6c2b4d79..94cdef9e 100644 --- a/src/ln/peer_handler.rs +++ b/src/ln/peer_handler.rs @@ -1,4 +1,5 @@ //! Top level peer message handling and socket handling logic lives here. +//! //! Instead of actually servicing sockets ourselves we require that you implement the //! SocketDescriptor interface and use that to receive actions which you should perform on the //! socket, and call into PeerManager with bytes read from the socket. The PeerManager will then @@ -32,7 +33,9 @@ pub struct MessageHandler { /// Provides an object which can be used to send data to and which uniquely identifies a connection /// to a remote host. You will need to be able to generate multiple of these which meet Eq and /// implement Hash to meet the PeerManager API. +/// /// For efficiency, Clone should be relatively cheap for this type. +/// /// You probably want to just extend an int and put a file descriptor in a struct and implement /// send_data. Note that if you are using a higher-level net library that may close() itself, be /// careful to ensure you don't have races whereby you might register a new connection with an fd @@ -42,9 +45,11 @@ pub trait SocketDescriptor : cmp::Eq + hash::Hash + Clone { /// Returns the amount of data which was sent, possibly 0 if the socket has since disconnected. /// Note that in the disconnected case, a disconnect_event must still fire and further write /// attempts may occur until that time. + /// /// If the returned size is smaller than data.len() - write_offset, a write_available event must /// trigger the next time more data can be written. Additionally, until the a send_data event /// completes fully, no further read_events should trigger on the same peer! + /// /// If a read_event on this descriptor had previously returned true (indicating that read /// events should be paused to prevent DoS in the send buffer), resume_read may be set /// indicating that read events on this descriptor should resume. A resume_read of false does @@ -167,6 +172,7 @@ impl PeerManager { } /// Get the list of node ids for peers which have completed the initial handshake. + /// /// For outbound connections, this will be the same as the their_node_id parameter passed in to /// new_outbound_connection, however entries will only appear once the initial handshake has /// completed and we are sure the remote peer has the private key for the given node_id. @@ -183,7 +189,9 @@ impl PeerManager { /// Indicates a new outbound connection has been established to a node with the given node_id. /// Note that if an Err is returned here you MUST NOT call disconnect_event for the new /// descriptor but must disconnect the connection immediately. + /// /// Returns some bytes to send to the remote node. + /// /// Panics if descriptor is duplicative with some other descriptor which has not yet has a /// disconnect_event. pub fn new_outbound_connection(&self, their_node_id: PublicKey, descriptor: Descriptor) -> Result, PeerHandleError> { @@ -213,10 +221,12 @@ impl PeerManager { } /// Indicates a new inbound connection has been established. + /// /// May refuse the connection by returning an Err, but will never write bytes to the remote end /// (outbound connector always speaks first). Note that if an Err is returned here you MUST NOT /// call disconnect_event for the new descriptor but must disconnect the connection /// immediately. + /// /// Panics if descriptor is duplicative with some other descriptor which has not yet has a /// disconnect_event. pub fn new_inbound_connection(&self, descriptor: Descriptor) -> Result<(), PeerHandleError> { @@ -266,12 +276,14 @@ impl PeerManager { } /// Indicates that there is room to write data to the given socket descriptor. + /// /// May return an Err to indicate that the connection should be closed. + /// /// Will most likely call send_data on the descriptor passed in (or the descriptor handed into - /// new_*_connection) before returning. Thus, be very careful with reentrancy issues! The + /// new_*\_connection) before returning. Thus, be very careful with reentrancy issues! The /// invariants around calling write_event in case a write did not fully complete must still /// hold - be ready to call write_event again if a write call generated here isn't sufficient! - /// Panics if the descriptor was not previously registered in a new_*_connection event. + /// Panics if the descriptor was not previously registered in a new_\*_connection event. pub fn write_event(&self, descriptor: &mut Descriptor) -> Result<(), PeerHandleError> { let mut peers = self.peers.lock().unwrap(); match peers.peers.get_mut(descriptor) { @@ -285,16 +297,20 @@ impl PeerManager { } /// Indicates that data was read from the given socket descriptor. + /// /// May return an Err to indicate that the connection should be closed. + /// /// Will very likely call send_data on the descriptor passed in (or a descriptor handed into /// new_*_connection) before returning. Thus, be very careful with reentrancy issues! The /// invariants around calling write_event in case a write did not fully complete must still /// hold. Note that this function will often call send_data on many peers before returning, not /// just this peer! + /// /// If Ok(true) is returned, further read_events should not be triggered until a write_event on /// this file descriptor has resume_read set (preventing DoS issues in the send buffer). Note /// that this must be true even if a send_data call with resume_read=true was made during the /// course of this function! + /// /// Panics if the descriptor was not previously registered in a new_*_connection event. pub fn read_event(&self, peer_descriptor: &mut Descriptor, data: Vec) -> Result { match self.do_read_event(peer_descriptor, data) { @@ -374,14 +390,12 @@ impl PeerManager { Ok(x) => x, Err(e) => { match e { - msgs::DecodeError::UnknownRealmByte => return Err(PeerHandleError{ no_connection_possible: false }), + msgs::DecodeError::UnknownVersion => 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::InvalidValue => 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"); @@ -389,7 +403,6 @@ impl PeerManager { }, msgs::DecodeError::BadLengthDescriptor => return Err(PeerHandleError{ no_connection_possible: false }), msgs::DecodeError::Io(_) => return Err(PeerHandleError{ no_connection_possible: false }), - msgs::DecodeError::InvalidValue => panic!("should not happen with message decoding"), } } }; @@ -602,10 +615,7 @@ impl PeerManager { }, 131 => { let msg = try_potential_decodeerror!(msgs::UpdateFailHTLC::read(&mut reader)); - let chan_update = try_potential_handleerror!(self.message_handler.chan_handler.handle_update_fail_htlc(&peer.their_node_id.unwrap(), &msg)); - if let Some(update) = chan_update { - self.message_handler.route_handler.handle_htlc_fail_channel_update(&update); - } + try_potential_handleerror!(self.message_handler.chan_handler.handle_update_fail_htlc(&peer.their_node_id.unwrap(), &msg)); }, 135 => { let msg = try_potential_decodeerror!(msgs::UpdateFailMalformedHTLC::read(&mut reader)); @@ -634,6 +644,9 @@ impl PeerManager { for resp in resps.update_fail_htlcs { encode_and_send_msg!(resp, 131); } + if let Some(resp) = resps.update_fee { + encode_and_send_msg!(resp, 134); + } encode_and_send_msg!(resps.commitment_signed, 132); }, None => {}, @@ -645,27 +658,44 @@ impl PeerManager { }, 136 => { let msg = try_potential_decodeerror!(msgs::ChannelReestablish::read(&mut reader)); - let (funding_locked, revoke_and_ack, commitment_update) = try_potential_handleerror!(self.message_handler.chan_handler.handle_channel_reestablish(&peer.their_node_id.unwrap(), &msg)); + let (funding_locked, revoke_and_ack, commitment_update, order) = try_potential_handleerror!(self.message_handler.chan_handler.handle_channel_reestablish(&peer.their_node_id.unwrap(), &msg)); if let Some(lock_msg) = funding_locked { encode_and_send_msg!(lock_msg, 36); } - if let Some(revoke_msg) = revoke_and_ack { - encode_and_send_msg!(revoke_msg, 133); - } - match commitment_update { - Some(resps) => { - for resp in resps.update_add_htlcs { - encode_and_send_msg!(resp, 128); - } - for resp in resps.update_fulfill_htlcs { - encode_and_send_msg!(resp, 130); - } - for resp in resps.update_fail_htlcs { - encode_and_send_msg!(resp, 131); - } - encode_and_send_msg!(resps.commitment_signed, 132); + macro_rules! handle_raa { () => { + if let Some(revoke_msg) = revoke_and_ack { + encode_and_send_msg!(revoke_msg, 133); + } + } } + macro_rules! handle_cu { () => { + match commitment_update { + Some(resps) => { + for resp in resps.update_add_htlcs { + encode_and_send_msg!(resp, 128); + } + for resp in resps.update_fulfill_htlcs { + encode_and_send_msg!(resp, 130); + } + for resp in resps.update_fail_htlcs { + encode_and_send_msg!(resp, 131); + } + if let Some(resp) = resps.update_fee { + encode_and_send_msg!(resp, 134); + } + encode_and_send_msg!(resps.commitment_signed, 132); + }, + None => {}, + } + } } + match order { + msgs::RAACommitmentOrder::RevokeAndACKFirst => { + handle_raa!(); + handle_cu!(); + }, + msgs::RAACommitmentOrder::CommitmentFirst => { + handle_cu!(); + handle_raa!(); }, - None => {}, } }, @@ -807,7 +837,7 @@ impl PeerManager { Self::do_attempt_write_data(&mut descriptor, peer); continue; }, - 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 } } => { + 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 update_fee, 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(), @@ -829,10 +859,24 @@ impl PeerManager { for msg in update_fail_malformed_htlcs { peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg, 135))); } + if let &Some(ref msg) = update_fee { + peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg, 134))); + } 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::SendRevokeAndACK { ref node_id, ref msg } => { + log_trace!(self, "Handling SendRevokeAndACK 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 + }); + peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg, 133))); + 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), @@ -884,6 +928,10 @@ impl PeerManager { } continue; }, + Event::PaymentFailureNetworkUpdate { ref update } => { + self.message_handler.route_handler.handle_htlc_fail_channel_update(update); + continue; + }, Event::HandleError { ref node_id, ref action } => { if let Some(ref action) = *action { match *action { @@ -938,8 +986,10 @@ impl PeerManager { } /// Indicates that the given socket descriptor's connection is now closed. + /// /// This must be called even if a PeerHandleError was given for a read_event or write_event, - /// but must NOT be called if a PeerHandleError was provided out of a new_*_connection event! + /// but must NOT be called if a PeerHandleError was provided out of a new_\*\_connection event! + /// /// Panics if the descriptor was not previously registered in a successful new_*_connection event. pub fn disconnect_event(&self, descriptor: &Descriptor) { self.disconnect_event_internal(descriptor, false);