Correct the "is peer live" checks in PeerManager

In general, we should be checking if a `Peer` has `their_features`
set as the "is this peer connected and have they finished the
handshake" flag as it indicates an `Init` message was received.

While none of these appear to be reachable bugs, there were a
number of places where we checked other flags for this purpose,
which may lead to sending messages before `Init` in the future.

Here we clean these cases up to always use the correct check (via
the new util method).

Author: TheBlueMatt

lightning/src/ln/peer_handler.rs

@@ -444,6 +444,7 @@ impl Peer {
 	/// point and we shouldn't send it yet to avoid sending duplicate updates. If we've already
 	/// sent the old versions, we should send the update, and so return true here.
 	fn should_forward_channel_announcement(&self, channel_id: u64) -> bool {
+		if self.their_features.is_none() { return false; }
 		if self.their_features.as_ref().unwrap().supports_gossip_queries() &&
 			!self.sent_gossip_timestamp_filter {
 				return false;
@@ -457,6 +458,7 @@ impl Peer {
 
 	/// Similar to the above, but for node announcements indexed by node_id.
 	fn should_forward_node_announcement(&self, node_id: NodeId) -> bool {
+		if self.their_features.is_none() { return false; }
 		if self.their_features.as_ref().unwrap().supports_gossip_queries() &&
 			!self.sent_gossip_timestamp_filter {
 				return false;
@@ -483,19 +485,20 @@ impl Peer {
 	fn should_buffer_gossip_backfill(&self) -> bool {
 		self.pending_outbound_buffer.is_empty() && self.gossip_broadcast_buffer.is_empty()
 			&& self.msgs_sent_since_pong < BUFFER_DRAIN_MSGS_PER_TICK
+			&& self.their_features.is_some()
 	}
 
 	/// Determines if we should push an onion message onto a peer's outbound buffer. This is checked
 	/// every time the peer's buffer may have been drained.
 	fn should_buffer_onion_message(&self) -> bool {
-		self.pending_outbound_buffer.is_empty()
+		self.pending_outbound_buffer.is_empty() && self.their_features.is_some()
 			&& self.msgs_sent_since_pong < BUFFER_DRAIN_MSGS_PER_TICK
 	}
 
 	/// Determines if we should push additional gossip broadcast messages onto a peer's outbound
 	/// buffer. This is checked every time the peer's buffer may have been drained.
 	fn should_buffer_gossip_broadcast(&self) -> bool {
-		self.pending_outbound_buffer.is_empty()
+		self.pending_outbound_buffer.is_empty() && self.their_features.is_some()
 			&& self.msgs_sent_since_pong < BUFFER_DRAIN_MSGS_PER_TICK
 	}
 
@@ -1537,10 +1540,12 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, OM: Deref, L: Deref, CM
 
 				for (_, peer_mutex) in peers.iter() {
 					let mut peer = peer_mutex.lock().unwrap();
-					if !peer.channel_encryptor.is_ready_for_encryption() || peer.their_features.is_none() ||
+					if peer.their_features.is_none() ||
 							!peer.should_forward_channel_announcement(msg.contents.short_channel_id) {
 						continue
 					}
+					debug_assert!(peer.their_node_id.is_some());
+					debug_assert!(peer.channel_encryptor.is_ready_for_encryption());
 					if peer.buffer_full_drop_gossip_broadcast() {
 						log_gossip!(self.logger, "Skipping broadcast message to {:?} as its outbound buffer is full", peer.their_node_id);
 						continue;
@@ -1562,10 +1567,12 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, OM: Deref, L: Deref, CM
 
 				for (_, peer_mutex) in peers.iter() {
 					let mut peer = peer_mutex.lock().unwrap();
-					if !peer.channel_encryptor.is_ready_for_encryption() || peer.their_features.is_none() ||
+					if peer.their_features.is_none() ||
 							!peer.should_forward_node_announcement(msg.contents.node_id) {
 						continue
 					}
+					debug_assert!(peer.their_node_id.is_some());
+					debug_assert!(peer.channel_encryptor.is_ready_for_encryption());
 					if peer.buffer_full_drop_gossip_broadcast() {
 						log_gossip!(self.logger, "Skipping broadcast message to {:?} as its outbound buffer is full", peer.their_node_id);
 						continue;
@@ -1587,10 +1594,12 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, OM: Deref, L: Deref, CM
 
 				for (_, peer_mutex) in peers.iter() {
 					let mut peer = peer_mutex.lock().unwrap();
-					if !peer.channel_encryptor.is_ready_for_encryption() || peer.their_features.is_none() ||
+					if peer.their_features.is_none() ||
 							!peer.should_forward_channel_announcement(msg.contents.short_channel_id)  {
 						continue
 					}
+					debug_assert!(peer.their_node_id.is_some());
+					debug_assert!(peer.channel_encryptor.is_ready_for_encryption());
 					if peer.buffer_full_drop_gossip_broadcast() {
 						log_gossip!(self.logger, "Skipping broadcast message to {:?} as its outbound buffer is full", peer.their_node_id);
 						continue;
@@ -2024,7 +2033,7 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, OM: Deref, L: Deref, CM
 				let mut peer = peer_mutex.lock().unwrap();
 				if flush_read_disabled { peer.received_channel_announce_since_backlogged = false; }
 
-				if !peer.channel_encryptor.is_ready_for_encryption() || peer.their_node_id.is_none() {
+				if peer.their_features.is_none() {
 					// The peer needs to complete its handshake before we can exchange messages. We
 					// give peers one timer tick to complete handshake, reusing
 					// `awaiting_pong_timer_tick_intervals` to track number of timer ticks taken
@@ -2036,6 +2045,8 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, OM: Deref, L: Deref, CM
 					}
 					continue;
 				}
+				debug_assert!(peer.channel_encryptor.is_ready_for_encryption());
+				debug_assert!(peer.their_node_id.is_some());
 
 				loop { // Used as a `goto` to skip writing a Ping message.
 					if peer.awaiting_pong_timer_tick_intervals == -1 {