Track # of outbound streams requested, not successfully established

The GossipSub protocol uses two streams, namely one from the local to the remote and one from the
remote to the local node. The rust-libp2p implementation enforces an upper limit on the churn of
both via `MAX_SUBSTREAM_CREATION`.

Thus far `self.outbound_substreams_created` was increased for every successful stream, but ignored
failed streams. Thus, on a given connection, a GossipSub handler could potentially indefinitely
retry creating an outbound stream, where each of those streams fails instead of succeeds to upgrade.

With this commit the handler tracks the number of requested outbound streams instead.

protocols/gossipsub/src/handler.rs

@@ -97,8 +97,8 @@ pub struct Handler {
     /// requests.
     outbound_substream_establishing: bool,
 
-    /// The number of outbound substreams we have created.
-    outbound_substreams_created: usize,
+    /// The number of outbound substreams we have requested.
+    outbound_substreams_requested: usize,
 
     /// The number of inbound substreams that have been created by the peer.
     inbound_substreams_created: usize,
@@ -161,7 +161,7 @@ impl Handler {
             inbound_substream: None,
             outbound_substream: None,
             outbound_substream_establishing: false,
-            outbound_substreams_created: 0,
+            outbound_substreams_requested: 0,
             inbound_substreams_created: 0,
             send_queue: SmallVec::new(),
             peer_kind: None,
@@ -216,8 +216,6 @@ impl Handler {
             return;
         }
 
-        self.outbound_substreams_created += 1;
-
         // update the known kind of peer
         if self.peer_kind.is_none() {
             self.peer_kind = Some(peer_kind);
@@ -311,6 +309,7 @@ impl ConnectionHandler for Handler {
             // Invariant: `self.outbound_substreams_created < MAX_SUBSTREAM_CREATION`.
             let message = self.send_queue.remove(0);
             self.send_queue.shrink_to_fit();
+            self.outbound_substreams_requested += 1;
             self.outbound_substream_establishing = true;
             return Poll::Ready(ConnectionHandlerEvent::OutboundSubstreamRequest {
                 protocol: self.listen_protocol.clone().map_info(|()| message),
@@ -516,7 +515,7 @@ impl ConnectionHandler for Handler {
             return;
         }
 
-        if event.is_outbound() && self.outbound_substreams_created == MAX_SUBSTREAM_CREATION {
+        if event.is_outbound() && self.outbound_substreams_requested == MAX_SUBSTREAM_CREATION {
             // Too many outbound substreams have been created, disable the handler.
             self.keep_alive = KeepAlive::No;
             log::info!("The maximum number of outbound substreams created has been exceeded.");