Give peers one timer tick to finish handshake before disconnecting

This ensures we don't let a hung connection stick around forever if
the peer never completes the initial handshake.

This also resolves a race where, on receiving a second connection
from a peer, we may reset their_node_id to None to prevent sending
messages even though the `channel_encryptor`
`is_ready_for_encryption()`. Sending pings only checks the
`channel_encryptor` status, not `their_node_id` resulting in an
`unwrap` on `None` in `enqueue_message`.

Author: TheBlueMatt

lightning/src/ln/peer_handler.rs

@@ -849,6 +849,7 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref, CMH: Deref> P
 									let features = InitFeatures::known();
 									let resp = msgs::Init { features };
 									self.enqueue_message(peer, &resp);
+									peer.awaiting_pong_tick_intervals = 0;
 								},
 								NextNoiseStep::ActThree => {
 									let their_node_id = try_potential_handleerror!(peer.channel_encryptor.process_act_three(&peer.pending_read_buffer[..]));
@@ -859,6 +860,7 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref, CMH: Deref> P
 									let features = InitFeatures::known();
 									let resp = msgs::Init { features };
 									self.enqueue_message(peer, &resp);
+									peer.awaiting_pong_tick_intervals = 0;
 								},
 								NextNoiseStep::NoiseComplete => {
 									if peer.pending_read_is_header {
@@ -1498,12 +1500,20 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref, CMH: Deref> P
 			let mut descriptors_needing_disconnect = Vec::new();
 
 			peers.retain(|descriptor, peer| {
-				if !peer.channel_encryptor.is_ready_for_encryption() {
-					// The peer needs to complete its handshake before we can exchange messages
-					return true;
+				let mut do_disconnect_peer = false;
+				if !peer.channel_encryptor.is_ready_for_encryption() || peer.their_node_id.is_none() {
+					// The peer needs to complete its handshake before we can exchange messages. We
+					// give peers one timer tick to complet handshake.
+					if peer.awaiting_pong_tick_intervals != 0 {
+						do_disconnect_peer = true;
+					} else {
+						peer.awaiting_pong_tick_intervals = 1;
+						return true;
+					}
 				}
 
-				if (peer.awaiting_pong_tick_intervals > 0 && !peer.received_message_since_timer_tick)
+				if do_disconnect_peer
+					|| (peer.awaiting_pong_tick_intervals > 0 && !peer.received_message_since_timer_tick)
 					|| peer.awaiting_pong_tick_intervals > MAX_BUFFER_DRAIN_TICK_INTERVALS
 				{
 					descriptors_needing_disconnect.push(descriptor.clone());
@@ -1513,11 +1523,7 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref, CMH: Deref> P
 							node_id_to_descriptor.remove(&node_id);
 							self.message_handler.chan_handler.peer_disconnected(&node_id, false);
 						}
-						None => {
-							// This can't actually happen as we should have hit
-							// is_ready_for_encryption() previously on this same peer.
-							unreachable!();
-						},
+						None => {},
 					}
 					return false;
 				}