Correct msgs_sent_since_pong tracking for gossip forwards

When we forward gossip messages, we use the already-encoded copy
we have, pushing it onto the `gossip_broadcast_buffer`. When we
have free socket buffer space, and there are no non-gossip messages
pending, we'll remove it from the `gossip_broadcast_buffer` and
push it onto the normal `pending_outbound_buffer` socket queue.

Separately, we use `msgs_sent_since_pong` to ensure that our peer
is being responsive and our messages are getting through with
minimal latency. After we send 32 messages (if we've already
received the `pong` for the last `ping` wensent), we send an extra
`ping` and will only queue up an additional 32 forwarded gossip
messages before we start dropping forwarded gossip on the floor.

Previously, we (arguably) incorrectly incremented
`msgs_sent_since_pong` when we pushed a message onto the
`gossip_broadcast_buffer`, not when it was actually queued up in
our `pending_outbound_buffer` immediate socket queue. This is
fairly strange - the point of `msgs_sent_since_pong` is to keep
peer latency low, we already independently enforce memory limits to
ensure we drop forwarded gossip messages if a peer's message queues
have grown too large. This means we potentially disconnect a peer
for not draining the backlog forwarded-gossip socket queue fast
enough, rather than simply dropping gossip.

While this shouldn't be a big deal in practice, I do see (mostly
Tor) peers disconnect on my node occasionally, and while its likely
due to Tor circuits hanging and a disconnect is needed, the
incorrect tracking here makes state analysis difficult as there are
nearly always a flow of enough gossip messages to "send to the
peer". Its also possible that this allows incredibly low bandwidth
connections to stay afloat more durably.

Author: TheBlueMatt

lightning/src/ln/peer_handler.rs

@@ -1551,6 +1551,7 @@ where
 			}
 			if peer.should_buffer_gossip_broadcast() {
 				if let Some(msg) = peer.gossip_broadcast_buffer.pop_front() {
+					peer.msgs_sent_since_pong += 1;
 					peer.pending_outbound_buffer
 						.push_back(peer.channel_encryptor.encrypt_buffer(msg));
 				}
@@ -1714,12 +1715,6 @@ where
 		peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(message));
 	}
 
-	/// Append a message to a peer's pending outbound/write gossip broadcast buffer
-	fn enqueue_encoded_gossip_broadcast(&self, peer: &mut Peer, encoded_message: MessageBuf) {
-		peer.msgs_sent_since_pong += 1;
-		peer.gossip_broadcast_buffer.push_back(encoded_message);
-	}
-
 	fn do_read_event(
 		&self, peer_descriptor: &mut Descriptor, data: &[u8],
 	) -> Result<bool, PeerHandleError> {
@@ -2689,10 +2684,8 @@ where
 					{
 						continue;
 					}
-					self.enqueue_encoded_gossip_broadcast(
-						&mut *peer,
-						MessageBuf::from_encoded(&encoded_msg),
-					);
+					let encoded_message = MessageBuf::from_encoded(&encoded_msg);
+					peer.gossip_broadcast_buffer.push_back(encoded_message);
 				}
 			},
 			wire::Message::NodeAnnouncement(ref msg) => {
@@ -2733,10 +2726,8 @@ where
 					{
 						continue;
 					}
-					self.enqueue_encoded_gossip_broadcast(
-						&mut *peer,
-						MessageBuf::from_encoded(&encoded_msg),
-					);
+					let encoded_message = MessageBuf::from_encoded(&encoded_msg);
+					peer.gossip_broadcast_buffer.push_back(encoded_message);
 				}
 			},
 			wire::Message::ChannelUpdate(ref msg) => {
@@ -2772,10 +2763,8 @@ where
 					{
 						continue;
 					}
-					self.enqueue_encoded_gossip_broadcast(
-						&mut *peer,
-						MessageBuf::from_encoded(&encoded_msg),
-					);
+					let encoded_message = MessageBuf::from_encoded(&encoded_msg);
+					peer.gossip_broadcast_buffer.push_back(encoded_message);
 				}
 			},
 			_ => {