Merge pull request #1374 from TheBlueMatt/2022-03-bindings-cleanups
[rust-lightning] / lightning / src / ln / peer_handler.rs
index 9157dc87dc2fc5e415de7679974bfe0de3a91f0c..eef4054b34ad794048dd3970e96471fdb6f931c2 100644 (file)
@@ -69,7 +69,7 @@ impl RoutingMessageHandler for IgnoringMessageHandler {
        fn get_next_channel_announcements(&self, _starting_point: u64, _batch_amount: u8) ->
                Vec<(msgs::ChannelAnnouncement, Option<msgs::ChannelUpdate>, Option<msgs::ChannelUpdate>)> { Vec::new() }
        fn get_next_node_announcements(&self, _starting_point: Option<&PublicKey>, _batch_amount: u8) -> Vec<msgs::NodeAnnouncement> { Vec::new() }
-       fn sync_routing_table(&self, _their_node_id: &PublicKey, _init: &msgs::Init) {}
+       fn peer_connected(&self, _their_node_id: &PublicKey, _init: &msgs::Init) {}
        fn handle_reply_channel_range(&self, _their_node_id: &PublicKey, _msg: msgs::ReplyChannelRange) -> Result<(), LightningError> { Ok(()) }
        fn handle_reply_short_channel_ids_end(&self, _their_node_id: &PublicKey, _msg: msgs::ReplyShortChannelIdsEnd) -> Result<(), LightningError> { Ok(()) }
        fn handle_query_channel_range(&self, _their_node_id: &PublicKey, _msg: msgs::QueryChannelRange) -> Result<(), LightningError> { Ok(()) }
@@ -307,12 +307,12 @@ const OUTBOUND_BUFFER_LIMIT_DROP_GOSSIP: usize = OUTBOUND_BUFFER_LIMIT_READ_PAUS
 /// Thus, to avoid needlessly disconnecting a peer, we allow a peer to take this many timer ticks
 /// per connected peer to respond to a ping, as long as they send us at least one message during
 /// each tick, ensuring we aren't actually just disconnected.
-/// With a timer tick interval of five seconds, this translates to about 30 seconds per connected
+/// With a timer tick interval of ten seconds, this translates to about 40 seconds per connected
 /// peer.
 ///
 /// When we improve parallelism somewhat we should reduce this to e.g. this many timer ticks per
 /// two connected peers, assuming most LDK-running systems have at least two cores.
-const MAX_BUFFER_DRAIN_TICK_INTERVALS_PER_PEER: i8 = 6;
+const MAX_BUFFER_DRAIN_TICK_INTERVALS_PER_PEER: i8 = 4;
 
 /// This is the minimum number of messages we expect a peer to be able to handle within one timer
 /// tick. Once we have sent this many messages since the last ping, we send a ping right away to
@@ -376,6 +376,8 @@ struct PeerHolder<Descriptor: SocketDescriptor> {
 /// lifetimes). Other times you can afford a reference, which is more efficient, in which case
 /// SimpleRefPeerManager is the more appropriate type. Defining these type aliases prevents
 /// issues such as overly long function definitions.
+///
+/// (C-not exported) as Arcs don't make sense in bindings
 pub type SimpleArcPeerManager<SD, M, T, F, C, L> = PeerManager<SD, Arc<SimpleArcChannelManager<M, T, F, L>>, Arc<NetGraphMsgHandler<Arc<NetworkGraph>, Arc<C>, Arc<L>>>, Arc<L>, Arc<IgnoringMessageHandler>>;
 
 /// SimpleRefPeerManager is a type alias for a PeerManager reference, and is the reference
@@ -384,6 +386,8 @@ pub type SimpleArcPeerManager<SD, M, T, F, C, L> = PeerManager<SD, Arc<SimpleArc
 /// usage of lightning-net-tokio (since tokio::spawn requires parameters with static lifetimes).
 /// But if this is not necessary, using a reference is more efficient. Defining these type aliases
 /// helps with issues such as long function definitions.
+///
+/// (C-not exported) as Arcs don't make sense in bindings
 pub type SimpleRefPeerManager<'a, 'b, 'c, 'd, 'e, 'f, 'g, 'h, SD, M, T, F, C, L> = PeerManager<SD, SimpleRefChannelManager<'a, 'b, 'c, 'd, 'e, M, T, F, L>, &'e NetGraphMsgHandler<&'g NetworkGraph, &'h C, &'f L>, &'f L, IgnoringMessageHandler>;
 
 /// A PeerManager manages a set of peers, described by their [`SocketDescriptor`] and marshalls
@@ -821,6 +825,11 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref, CMH: Deref> P
                                                                                                        self.enqueue_message(peer, &msg);
                                                                                                        continue;
                                                                                                },
+                                                                                               msgs::ErrorAction::SendWarningMessage { msg, log_level } => {
+                                                                                                       log_given_level!(self.logger, log_level, "Error handling message{}; sending warning message with: {}", OptionalFromDebugger(&peer.their_node_id), e.err);
+                                                                                                       self.enqueue_message(peer, &msg);
+                                                                                                       continue;
+                                                                                               },
                                                                                        }
                                                                                }
                                                                        }
@@ -897,25 +906,40 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref, CMH: Deref> P
                                                                                        Ok(x) => x,
                                                                                        Err(e) => {
                                                                                                match e {
-                                                                                                       msgs::DecodeError::UnknownVersion => return Err(PeerHandleError { no_connection_possible: false }),
-                                                                                                       msgs::DecodeError::UnknownRequiredFeature => {
+                                                                                                       // Note that to avoid recursion we never call
+                                                                                                       // `do_attempt_write_data` from here, causing
+                                                                                                       // the messages enqueued here to not actually
+                                                                                                       // be sent before the peer is disconnected.
+                                                                                                       (msgs::DecodeError::UnknownRequiredFeature, Some(ty)) if is_gossip_msg(ty) => {
                                                                                                                log_gossip!(self.logger, "Got a channel/node announcement with an unknown required feature flag, you may want to update!");
                                                                                                                continue;
                                                                                                        }
-                                                                                                       msgs::DecodeError::InvalidValue => {
+                                                                                                       (msgs::DecodeError::UnsupportedCompression, _) => {
+                                                                                                               log_gossip!(self.logger, "We don't support zlib-compressed message fields, sending a warning and ignoring message");
+                                                                                                               self.enqueue_message(peer, &msgs::WarningMessage { channel_id: [0; 32], data: "Unsupported message compression: zlib".to_owned() });
+                                                                                                               continue;
+                                                                                                       }
+                                                                                                       (_, Some(ty)) if is_gossip_msg(ty) => {
+                                                                                                               log_gossip!(self.logger, "Got an invalid value while deserializing a gossip message");
+                                                                                                               self.enqueue_message(peer, &msgs::WarningMessage { channel_id: [0; 32], data: "Unreadable/bogus gossip message".to_owned() });
+                                                                                                               continue;
+                                                                                                       }
+                                                                                                       (msgs::DecodeError::UnknownRequiredFeature, ty) => {
+                                                                                                               log_gossip!(self.logger, "Received a message with an unknown required feature flag or TLV, you may want to update!");
+                                                                                                               self.enqueue_message(peer, &msgs::WarningMessage { channel_id: [0; 32], data: format!("Received an unknown required feature/TLV in message type {:?}", ty) });
+                                                                                                               return Err(PeerHandleError { no_connection_possible: false });
+                                                                                                       }
+                                                                                                       (msgs::DecodeError::UnknownVersion, _) => return Err(PeerHandleError { no_connection_possible: false }),
+                                                                                                       (msgs::DecodeError::InvalidValue, _) => {
                                                                                                                log_debug!(self.logger, "Got an invalid value while deserializing message");
                                                                                                                return Err(PeerHandleError { no_connection_possible: false });
                                                                                                        }
-                                                                                                       msgs::DecodeError::ShortRead => {
+                                                                                                       (msgs::DecodeError::ShortRead, _) => {
                                                                                                                log_debug!(self.logger, "Deserialization failed due to shortness of message");
                                                                                                                return Err(PeerHandleError { no_connection_possible: false });
                                                                                                        }
-                                                                                                       msgs::DecodeError::BadLengthDescriptor => return Err(PeerHandleError { no_connection_possible: false }),
-                                                                                                       msgs::DecodeError::Io(_) => return Err(PeerHandleError { no_connection_possible: false }),
-                                                                                                       msgs::DecodeError::UnsupportedCompression => {
-                                                                                                               log_gossip!(self.logger, "We don't support zlib-compressed message fields, ignoring message");
-                                                                                                               continue;
-                                                                                                       }
+                                                                                                       (msgs::DecodeError::BadLengthDescriptor, _) => return Err(PeerHandleError { no_connection_possible: false }),
+                                                                                                       (msgs::DecodeError::Io(_), _) => return Err(PeerHandleError { no_connection_possible: false }),
                                                                                                }
                                                                                        }
                                                                                };
@@ -998,7 +1022,7 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref, CMH: Deref> P
                                        return Err(PeerHandleError{ no_connection_possible: true }.into());
                                }
 
-                               self.message_handler.route_handler.sync_routing_table(&peer.their_node_id.unwrap(), &msg);
+                               self.message_handler.route_handler.peer_connected(&peer.their_node_id.unwrap(), &msg);
 
                                self.message_handler.chan_handler.peer_connected(&peer.their_node_id.unwrap(), &msg);
                                peer.their_features = Some(msg.features);
@@ -1022,6 +1046,21 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref, CMH: Deref> P
                                        return Err(PeerHandleError{ no_connection_possible: true }.into());
                                }
                        },
+                       wire::Message::Warning(msg) => {
+                               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.logger, "Got warning message from {}: {}", log_pubkey!(peer.their_node_id.unwrap()), msg.data);
+                               } else {
+                                       log_debug!(self.logger, "Got warning message from {} with non-ASCII error message", log_pubkey!(peer.their_node_id.unwrap()));
+                               }
+                       },
 
                        wire::Message::Ping(msg) => {
                                if msg.ponglen < 65532 {
@@ -1419,6 +1458,12 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref, CMH: Deref> P
                                                                                msg.data);
                                                                self.enqueue_message(get_peer_for_forwarding!(node_id), msg);
                                                        },
+                                                       msgs::ErrorAction::SendWarningMessage { ref msg, ref log_level } => {
+                                                               log_given_level!(self.logger, *log_level, "Handling SendWarningMessage HandleError event in peer_handler for node {} with message {}",
+                                                                               log_pubkey!(node_id),
+                                                                               msg.data);
+                                                               self.enqueue_message(get_peer_for_forwarding!(node_id), msg);
+                                                       },
                                                }
                                        },
                                        MessageSendEvent::SendChannelRangeQuery { ref node_id, ref msg } => {
@@ -1436,6 +1481,9 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref, CMH: Deref> P
                                                        msg.sync_complete);
                                                self.enqueue_message(get_peer_for_forwarding!(node_id), msg);
                                        }
+                                       MessageSendEvent::SendGossipTimestampFilter { ref node_id, ref msg } => {
+                                               self.enqueue_message(get_peer_for_forwarding!(node_id), msg);
+                                       }
                                }
                        }
 
@@ -1531,9 +1579,9 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref, CMH: Deref> P
        /// Send pings to each peer and disconnect those which did not respond to the last round of
        /// pings.
        ///
-       /// This may be called on any timescale you want, however, roughly once every five to ten
-       /// seconds is preferred. The call rate determines both how often we send a ping to our peers
-       /// and how much time they have to respond before we disconnect them.
+       /// This may be called on any timescale you want, however, roughly once every ten seconds is
+       /// preferred. The call rate determines both how often we send a ping to our peers and how much
+       /// time they have to respond before we disconnect them.
        ///
        /// May call [`send_data`] on all [`SocketDescriptor`]s. Thus, be very careful with reentrancy
        /// issues!